QPID-2815: refactor broker startup to present a clean interface interface for starting the broker within an existing application

Applied patch by Keith Wall and myself

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

27 files changed, 1348 insertions, 744 deletions

diff --git a/java/broker-plugins/firewall/src/test/java/org/apache/qpid/server/security/access/FirewallConfigurationTest.java b/java/broker-plugins/firewall/src/test/java/org/apache/qpid/server/security/access/FirewallConfigurationTest.java
index d230a49bde..24b8bbe36c 100644
--- a/java/broker-plugins/firewall/src/test/java/org/apache/qpid/server/security/access/FirewallConfigurationTest.java
+++ b/java/broker-plugins/firewall/src/test/java/org/apache/qpid/server/security/access/FirewallConfigurationTest.java
@@ -34,6 +34,13 @@ import org.apache.qpid.test.utils.QpidTestCase;
 
 public class FirewallConfigurationTest extends QpidTestCase
 {
+    @Override
+    protected void tearDown() throws Exception
+    {
+        super.tearDown();
+        ApplicationRegistry.remove();
+    }
+
     public void testFirewallConfiguration() throws Exception
     {
         // Write out config
@@ -43,18 +50,11 @@ public class FirewallConfigurationTest extends QpidTestCase
 
         // Load config
         ApplicationRegistry reg = new ConfigurationFileApplicationRegistry(mainFile);
-        try
-        {
-            ApplicationRegistry.initialise(reg);
+        ApplicationRegistry.initialise(reg);
 
-            // Test config
-            assertFalse(reg.getSecurityManager().accessVirtualhost("test", new InetSocketAddress("127.0.0.1", 65535)));
-            assertTrue(reg.getSecurityManager().accessVirtualhost("test", new InetSocketAddress("127.1.2.3", 65535)));
-        }
-        finally
-        {
-            ApplicationRegistry.remove();
-        }
+        // Test config
+        assertFalse(reg.getSecurityManager().accessVirtualhost("test", new InetSocketAddress("127.0.0.1", 65535)));
+        assertTrue(reg.getSecurityManager().accessVirtualhost("test", new InetSocketAddress("127.1.2.3", 65535)));
     }
 
     public void testCombinedConfigurationFirewall() throws Exception
@@ -112,17 +112,10 @@ public class FirewallConfigurationTest extends QpidTestCase
 
         // Load config
         ApplicationRegistry reg = new ConfigurationFileApplicationRegistry(mainFile);
-        try
-        {
-            ApplicationRegistry.initialise(reg);
+        ApplicationRegistry.initialise(reg);
 
-            // Test config
-            assertFalse(reg.getSecurityManager().accessVirtualhost("test", new InetSocketAddress("127.0.0.1", 65535)));
-        }
-        finally
-        {
-            ApplicationRegistry.remove();
-        }
+        // Test config
+        assertFalse(reg.getSecurityManager().accessVirtualhost("test", new InetSocketAddress("127.0.0.1", 65535)));
     }
 
     public void testConfigurationFirewallReload() throws Exception
@@ -135,24 +128,17 @@ public class FirewallConfigurationTest extends QpidTestCase
 
         // Load config
         ApplicationRegistry reg = new ConfigurationFileApplicationRegistry(mainFile);
-        try
-        {
-            ApplicationRegistry.initialise(reg);
+        ApplicationRegistry.initialise(reg);
 
-            // Test config
-            assertFalse(reg.getSecurityManager().accessVirtualhost("test", new InetSocketAddress("127.0.0.1", 65535)));
+        // Test config
+        assertFalse(reg.getSecurityManager().accessVirtualhost("test", new InetSocketAddress("127.0.0.1", 65535)));
 
-            // Switch to deny the connection
-            writeConfigFile(mainFile, true);
+        // Switch to deny the connection
+        writeConfigFile(mainFile, true);
 
-            reg.getConfiguration().reparseConfigFileSecuritySections();
+        reg.getConfiguration().reparseConfigFileSecuritySections();
 
-            assertTrue(reg.getSecurityManager().accessVirtualhost("test", new InetSocketAddress("127.0.0.1", 65535)));
-        }
-        finally
-        {
-            ApplicationRegistry.remove();
-        }
+        assertTrue(reg.getSecurityManager().accessVirtualhost("test", new InetSocketAddress("127.0.0.1", 65535)));
     }
 
     public void testCombinedConfigurationFirewallReload() throws Exception
@@ -209,47 +195,40 @@ public class FirewallConfigurationTest extends QpidTestCase
 
         // Load config
         ApplicationRegistry reg = new ConfigurationFileApplicationRegistry(mainFile);
-        try
-        {
-            ApplicationRegistry.initialise(reg);
+        ApplicationRegistry.initialise(reg);
 
-            // Test config
-            assertFalse(reg.getSecurityManager().accessVirtualhost("test", new InetSocketAddress("127.0.0.1", 65535)));
+        // Test config
+        assertFalse(reg.getSecurityManager().accessVirtualhost("test", new InetSocketAddress("127.0.0.1", 65535)));
 
-            RandomAccessFile fileBRandom = new RandomAccessFile(fileB, "rw");
-            fileBRandom.setLength(0);
-            fileBRandom.seek(0);
-            fileBRandom.close();
+        RandomAccessFile fileBRandom = new RandomAccessFile(fileB, "rw");
+        fileBRandom.setLength(0);
+        fileBRandom.seek(0);
+        fileBRandom.close();
 
-            out = new FileWriter(fileB);
-            out.write("<firewall>\n");
-            out.write("\t<rule access=\"allow\" network=\"127.0.0.1\"/>");
-            out.write("</firewall>\n");
-            out.close();
+        out = new FileWriter(fileB);
+        out.write("<firewall>\n");
+        out.write("\t<rule access=\"allow\" network=\"127.0.0.1\"/>");
+        out.write("</firewall>\n");
+        out.close();
 
-            reg.getConfiguration().reparseConfigFileSecuritySections();
+        reg.getConfiguration().reparseConfigFileSecuritySections();
 
-            assertTrue(reg.getSecurityManager().accessVirtualhost("test", new InetSocketAddress("127.0.0.1", 65535)));
+        assertTrue(reg.getSecurityManager().accessVirtualhost("test", new InetSocketAddress("127.0.0.1", 65535)));
 
-            fileBRandom = new RandomAccessFile(fileB, "rw");
-            fileBRandom.setLength(0);
-            fileBRandom.seek(0);
-            fileBRandom.close();
+        fileBRandom = new RandomAccessFile(fileB, "rw");
+        fileBRandom.setLength(0);
+        fileBRandom.seek(0);
+        fileBRandom.close();
 
-            out = new FileWriter(fileB);
-            out.write("<firewall>\n");
-            out.write("\t<rule access=\"deny\" network=\"127.0.0.1\"/>");
-            out.write("</firewall>\n");
-            out.close();
+        out = new FileWriter(fileB);
+        out.write("<firewall>\n");
+        out.write("\t<rule access=\"deny\" network=\"127.0.0.1\"/>");
+        out.write("</firewall>\n");
+        out.close();
 
-            reg.getConfiguration().reparseConfigFileSecuritySections();
+        reg.getConfiguration().reparseConfigFileSecuritySections();
 
-            assertFalse(reg.getSecurityManager().accessVirtualhost("test", new InetSocketAddress("127.0.0.1", 65535)));
-        }
-        finally
-        {
-            ApplicationRegistry.remove();
-        }
+        assertFalse(reg.getSecurityManager().accessVirtualhost("test", new InetSocketAddress("127.0.0.1", 65535)));
     }
 
     private void writeFirewallVhostsFile(File vhostsFile, boolean allow) throws IOException
@@ -335,7 +314,7 @@ public class FirewallConfigurationTest extends QpidTestCase
         // Load config
         ApplicationRegistry reg = new ConfigurationFileApplicationRegistry(mainFile);
         ApplicationRegistry.initialise(reg);
-
+    
         // Test config
         VirtualHostRegistry virtualHostRegistry = reg.getVirtualHostRegistry();
         VirtualHost virtualHost = virtualHostRegistry.getVirtualHost("test");
diff --git a/java/broker-plugins/firewall/src/test/java/org/apache/qpid/server/security/access/FirewallPluginTest.java b/java/broker-plugins/firewall/src/test/java/org/apache/qpid/server/security/access/FirewallPluginTest.java
index 2b04962c89..00077d9d9c 100644
--- a/java/broker-plugins/firewall/src/test/java/org/apache/qpid/server/security/access/FirewallPluginTest.java
+++ b/java/broker-plugins/firewall/src/test/java/org/apache/qpid/server/security/access/FirewallPluginTest.java
@@ -27,12 +27,15 @@ import java.net.SocketAddress;
 
 import org.apache.commons.configuration.ConfigurationException;
 import org.apache.commons.configuration.XMLConfiguration;
+import org.apache.qpid.server.configuration.ServerConfiguration;
+import org.apache.qpid.server.registry.ApplicationRegistry;
 import org.apache.qpid.server.security.Result;
 import org.apache.qpid.server.security.access.plugins.Firewall;
 import org.apache.qpid.server.security.access.plugins.FirewallConfiguration;
-import org.apache.qpid.server.util.InternalBrokerBaseCase;
+import org.apache.qpid.server.util.TestApplicationRegistry;
+import org.apache.qpid.test.utils.QpidTestCase;
 
-public class FirewallPluginTest extends InternalBrokerBaseCase
+public class FirewallPluginTest extends QpidTestCase
 {
     public class RuleInfo
     {
@@ -73,15 +76,23 @@ public class FirewallPluginTest extends InternalBrokerBaseCase
 
     // IP address
     private SocketAddress _address;
+    private ServerConfiguration _serverConfig;
 
     @Override
-    public void setUp() throws Exception
+    protected void setUp() throws Exception
     {
         super.setUp();
-        
+        _serverConfig = new ServerConfiguration(new XMLConfiguration());
+        ApplicationRegistry.initialise(new TestApplicationRegistry(_serverConfig));
         _address = new InetSocketAddress("127.0.0.1", 65535);
     }
 
+    @Override
+    protected void tearDown() throws Exception
+    {
+        super.tearDown();
+        ApplicationRegistry.remove();
+    }
     private Firewall initialisePlugin(String defaultAction, RuleInfo[] rules) throws IOException, ConfigurationException
     {
         // Create sample config file
@@ -108,7 +119,7 @@ public class FirewallPluginTest extends InternalBrokerBaseCase
         }
         buf.write("</firewall>");
         buf.close();
-        
+
         // Configure plugin
         FirewallConfiguration config = new FirewallConfiguration();
         config.setConfiguration("", new XMLConfiguration(confFile));
diff --git a/java/broker/src/main/java/org/apache/qpid/server/Broker.java b/java/broker/src/main/java/org/apache/qpid/server/Broker.java
new file mode 100644
index 0000000000..124c2a7d08
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/Broker.java
@@ -0,0 +1,420 @@
+/*
+ *
+ * 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;
+
+import static org.apache.qpid.transport.ConnectionSettings.WILDCARD_ADDRESS;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.util.EnumSet;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+
+import org.apache.log4j.PropertyConfigurator;
+import org.apache.log4j.xml.QpidLog4JConfigurator;
+import org.apache.qpid.server.configuration.ServerConfiguration;
+import org.apache.qpid.server.configuration.ServerNetworkTransportConfiguration;
+import org.apache.qpid.server.configuration.management.ConfigurationManagementMBean;
+import org.apache.qpid.server.information.management.ServerInformationMBean;
+import org.apache.qpid.server.logging.SystemOutMessageLogger;
+import org.apache.qpid.server.logging.actors.BrokerActor;
+import org.apache.qpid.server.logging.actors.CurrentActor;
+import org.apache.qpid.server.logging.actors.GenericActor;
+import org.apache.qpid.server.logging.management.LoggingManagementMBean;
+import org.apache.qpid.server.logging.messages.BrokerMessages;
+import org.apache.qpid.server.protocol.AMQProtocolEngineFactory;
+import org.apache.qpid.server.protocol.MultiVersionProtocolEngineFactory;
+import org.apache.qpid.server.protocol.MultiVersionProtocolEngineFactory.VERSION;
+import org.apache.qpid.server.registry.ApplicationRegistry;
+import org.apache.qpid.server.registry.ConfigurationFileApplicationRegistry;
+import org.apache.qpid.server.transport.QpidAcceptor;
+import org.apache.qpid.ssl.SSLContextFactory;
+import org.apache.qpid.transport.NetworkTransportConfiguration;
+import org.apache.qpid.transport.network.IncomingNetworkTransport;
+import org.apache.qpid.transport.network.Transport;
+import org.apache.qpid.transport.network.mina.MinaNetworkTransport;
+
+public class Broker
+{
+    private static final int IPV4_ADDRESS_LENGTH = 4;
+    private static final char IPV4_LITERAL_SEPARATOR = '.';
+
+    protected static class InitException extends RuntimeException
+    {
+        private static final long serialVersionUID = 1L;
+
+        InitException(String msg, Throwable cause)
+        {
+            super(msg, cause);
+        }
+    }
+
+    public void shutdown()
+    {
+        ApplicationRegistry.remove();
+    }
+
+    public void startup() throws Exception
+    {
+        startup(new BrokerOptions());
+    }
+
+    public void startup(BrokerOptions options) throws Exception
+    {
+        try
+        {
+            CurrentActor.set(new BrokerActor(new SystemOutMessageLogger()));
+            startupImpl(options);
+        }
+        finally
+        {
+            CurrentActor.remove();
+        }
+    }
+
+    private void startupImpl(final BrokerOptions options) throws Exception
+    {
+        final String qpidHome = options.getQpidHome();
+        final File configFile = getConfigFile(options.getConfigFile(),
+                                    BrokerOptions.DEFAULT_CONFIG_FILE, qpidHome, true);
+
+        CurrentActor.get().message(BrokerMessages.CONFIG(configFile.getAbsolutePath()));
+
+        File logConfigFile = getConfigFile(options.getLogConfigFile(),
+                                    BrokerOptions.DEFAULT_LOG_CONFIG_FILE, qpidHome, false);
+
+        configureLogging(logConfigFile, options.getLogWatchFrequency());
+
+        ConfigurationFileApplicationRegistry config = new ConfigurationFileApplicationRegistry(configFile);
+        ServerConfiguration serverConfig = config.getConfiguration();
+        updateManagementPort(serverConfig, options.getJmxPort());
+
+        ApplicationRegistry.initialise(config);
+
+        // We have already loaded the BrokerMessages class by this point so we
+        // need to refresh the locale setting incase we had a different value in
+        // the configuration.
+        BrokerMessages.reload();
+
+        // AR.initialise() sets and removes its own actor so we now need to set the actor
+        // for the remainder of the startup, and the default actor if the stack is empty
+        CurrentActor.set(new BrokerActor(config.getCompositeStartupMessageLogger()));
+        CurrentActor.setDefault(new BrokerActor(config.getRootMessageLogger()));
+        GenericActor.setDefaultMessageLogger(config.getRootMessageLogger());
+
+        try
+        {
+            configureLoggingManagementMBean(logConfigFile, options.getLogWatchFrequency());
+
+            ConfigurationManagementMBean configMBean = new ConfigurationManagementMBean();
+            configMBean.register();
+
+            ServerInformationMBean sysInfoMBean = new ServerInformationMBean(config);
+            sysInfoMBean.register();
+
+            Set<Integer> ports = new HashSet<Integer>(options.getPorts());
+            if(ports.isEmpty())
+            {
+                parsePortList(ports, serverConfig.getPorts());
+            }
+
+            Set<Integer> sslPorts = new HashSet<Integer>(options.getSSLPorts());
+            if(sslPorts.isEmpty())
+            {
+                parsePortList(sslPorts, serverConfig.getSSLPorts());
+            }
+
+            Set<Integer> exclude_0_10 = new HashSet<Integer>(options.getExcludedPorts(ProtocolExclusion.v0_10));
+            if(exclude_0_10.isEmpty())
+            {
+                parsePortList(exclude_0_10, serverConfig.getPortExclude010());
+            }
+
+            Set<Integer> exclude_0_9_1 = new HashSet<Integer>(options.getExcludedPorts(ProtocolExclusion.v0_9_1));
+            if(exclude_0_9_1.isEmpty())
+            {
+                parsePortList(exclude_0_9_1, serverConfig.getPortExclude091());
+            }
+
+            Set<Integer> exclude_0_9 = new HashSet<Integer>(options.getExcludedPorts(ProtocolExclusion.v0_9));
+            if(exclude_0_9.isEmpty())
+            {
+                parsePortList(exclude_0_9, serverConfig.getPortExclude09());
+            }
+
+            Set<Integer> exclude_0_8 = new HashSet<Integer>(options.getExcludedPorts(ProtocolExclusion.v0_8));
+            if(exclude_0_8.isEmpty())
+            {
+                parsePortList(exclude_0_8, serverConfig.getPortExclude08());
+            }
+
+            String bindAddr = options.getBind();
+            if (bindAddr == null)
+            {
+                bindAddr = serverConfig.getBind();
+            }
+
+            InetAddress bindAddress = null;
+            if (bindAddr.equals(WILDCARD_ADDRESS))
+            {
+                bindAddress = new InetSocketAddress(0).getAddress();
+            }
+            else
+            {
+                bindAddress = InetAddress.getByAddress(parseIP(bindAddr));
+            }
+            String hostName = bindAddress.getCanonicalHostName();
+
+            if (!serverConfig.getSSLOnly())
+            {
+                for(int port : ports)
+                {
+                    Set<VERSION> supported = EnumSet.allOf(VERSION.class);
+
+                    if(exclude_0_10.contains(port))
+                    {
+                        supported.remove(VERSION.v0_10);
+                    }
+
+                    if(exclude_0_9_1.contains(port))
+                    {
+                        supported.remove(VERSION.v0_9_1);
+                    }
+                    if(exclude_0_9.contains(port))
+                    {
+                        supported.remove(VERSION.v0_9);
+                    }
+                    if(exclude_0_8.contains(port))
+                    {
+                        supported.remove(VERSION.v0_8);
+                    }
+
+                    NetworkTransportConfiguration settings = 
+                        new ServerNetworkTransportConfiguration(serverConfig, port, bindAddress.getHostName(), Transport.TCP);
+
+                    IncomingNetworkTransport transport = new MinaNetworkTransport();
+                    MultiVersionProtocolEngineFactory protocolEngineFactory =
+                        new MultiVersionProtocolEngineFactory(hostName, supported);
+
+                    transport.accept(settings, protocolEngineFactory, null);
+                    ApplicationRegistry.getInstance().addAcceptor(new InetSocketAddress(bindAddress, port),
+                            new QpidAcceptor(transport,"TCP"));
+                    CurrentActor.get().message(BrokerMessages.LISTENING("TCP", port));
+                }
+            }
+
+            if (serverConfig.getEnableSSL())
+            {
+                String keystorePath = serverConfig.getKeystorePath();
+                String keystorePassword = serverConfig.getKeystorePassword();
+                String certType = serverConfig.getCertType();
+                SSLContextFactory sslFactory = 
+                    new SSLContextFactory(keystorePath, keystorePassword, certType);
+
+                for(int sslPort : sslPorts)
+                {
+                    NetworkTransportConfiguration settings = 
+                        new ServerNetworkTransportConfiguration(serverConfig, sslPort, bindAddress.getHostName(), Transport.TCP);
+
+                    IncomingNetworkTransport transport = new MinaNetworkTransport();
+
+                    transport.accept(settings, new AMQProtocolEngineFactory(), sslFactory);
+
+                    ApplicationRegistry.getInstance().addAcceptor(new InetSocketAddress(bindAddress, sslPort),
+                            new QpidAcceptor(transport,"TCP"));
+                    CurrentActor.get().message(BrokerMessages.LISTENING("TCP/SSL", sslPort));
+                }
+            }
+
+            CurrentActor.get().message(BrokerMessages.READY());
+        }
+        finally
+        {
+            // Startup is complete so remove the AR initialised Startup actor
+            CurrentActor.remove();
+        }
+    }
+
+    private File getConfigFile(final String fileName,
+                               final String defaultFileName,
+                               final String qpidHome, boolean throwOnFileNotFound) throws InitException
+    {
+        File configFile = null;
+        if (fileName != null)
+        {
+            configFile = new File(fileName);
+        }
+        else
+        {
+            configFile = new File(qpidHome, defaultFileName);
+        }
+
+        if (!configFile.exists() && throwOnFileNotFound)
+        {
+            String error = "File " + fileName + " could not be found. Check the file exists and is readable.";
+
+            if (qpidHome == null)
+            {
+                error = error + "\nNote: " + BrokerOptions.QPID_HOME + " is not set.";
+            }
+
+            throw new InitException(error, null);
+        }
+
+        return configFile;
+    }
+
+    public static void parsePortList(Set<Integer> output, List<?> ports) throws InitException
+    {
+        if(ports != null)
+        {
+            for(Object o : ports)
+            {
+                try
+                {
+                    output.add(Integer.parseInt(String.valueOf(o)));
+                }
+                catch (NumberFormatException e)
+                {
+                    throw new InitException("Invalid port: " + o, e);
+                }
+            }
+        }
+    }
+
+    /**
+     * Update the configuration data with the management port.
+     * @param configuration
+     * @param managementPort The string from the command line
+     */
+    private void updateManagementPort(ServerConfiguration configuration, Integer managementPort)
+    {
+        if (managementPort != null)
+        {
+            try
+            {
+                configuration.setJMXManagementPort(managementPort);
+            }
+            catch (NumberFormatException e)
+            {
+                throw new InitException("Invalid management port: " + managementPort, null);
+            }
+        }
+    }
+
+    private byte[] parseIP(String address) throws Exception
+    {
+        char[] literalBuffer = address.toCharArray();
+        int byteCount = 0;
+        int currByte = 0;
+        byte[] ip = new byte[IPV4_ADDRESS_LENGTH];
+        for (int i = 0; i < literalBuffer.length; i++)
+        {
+            char currChar = literalBuffer[i];
+            if ((currChar >= '0') && (currChar <= '9'))
+            {
+                currByte = (currByte * 10) + (Character.digit(currChar, 10) & 0xFF);
+            }
+
+            if (currChar == IPV4_LITERAL_SEPARATOR || (i + 1 == literalBuffer.length))
+            {
+                ip[byteCount++] = (byte) currByte;
+                currByte = 0;
+            }
+        }
+
+        if (byteCount != 4)
+        {
+            throw new Exception("Invalid IP address: " + address);
+        }
+        return ip;
+    }
+
+    private void configureLogging(File logConfigFile, long logWatchTime) throws InitException, IOException
+    {
+        if (logConfigFile.exists() && logConfigFile.canRead())
+        {
+            CurrentActor.get().message(BrokerMessages.LOG_CONFIG(logConfigFile.getAbsolutePath()));
+
+            if (logWatchTime > 0)
+            {
+                System.out.println("log file " + logConfigFile.getAbsolutePath() + " will be checked for changes every "
+                        + logWatchTime + " seconds");
+                // log4j expects the watch interval in milliseconds
+                try
+                {
+                    QpidLog4JConfigurator.configureAndWatch(logConfigFile.getPath(), logWatchTime);
+                }
+                catch (Exception e)
+                {
+                    throw new InitException(e.getMessage(),e);
+                }
+            }
+            else
+            {
+                try
+                {
+                    QpidLog4JConfigurator.configure(logConfigFile.getPath());
+                }
+                catch (Exception e)
+                {
+                    throw new InitException(e.getMessage(),e);
+                }
+            }
+        }
+        else
+        {
+            System.err.println("Logging configuration error: unable to read file " + logConfigFile.getAbsolutePath());
+            System.err.println("Using the fallback internal log4j.properties configuration");
+
+            InputStream propsFile = this.getClass().getResourceAsStream("/log4j.properties");
+            if(propsFile == null)
+            {
+                throw new IOException("Unable to load the fallback internal log4j.properties configuration file");
+            }
+            else
+            {
+                try
+                {
+                    Properties fallbackProps = new Properties();
+                    fallbackProps.load(propsFile);
+                    PropertyConfigurator.configure(fallbackProps);
+                }
+                finally
+                {
+                    propsFile.close();
+                }
+            }
+        }
+    }
+
+    private void configureLoggingManagementMBean(File logConfigFile, int logWatchTime) throws Exception
+    {
+        LoggingManagementMBean blm = new LoggingManagementMBean(logConfigFile.getPath(),logWatchTime);
+
+        blm.register();
+    }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/BrokerOptions.java b/java/broker/src/main/java/org/apache/qpid/server/BrokerOptions.java
new file mode 100644
index 0000000000..bf554c526f
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/BrokerOptions.java
@@ -0,0 +1,160 @@
+/*
+ *
+ * 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;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+public class BrokerOptions
+{
+    /** serialVersionUID */
+    private static final long serialVersionUID = 8051825964945442234L;
+
+    public static final String DEFAULT_CONFIG_FILE = "etc/config.xml";
+    public static final String DEFAULT_LOG_CONFIG_FILE = "etc/log4j.xml";
+    public static final String QPID_HOME = "QPID_HOME";
+
+    public static final String PORTS = "p";
+    public static final String SSL_PORTS = "s";
+    public static final String BIND = "b";
+    public static final String MANAGEMENT = "m";
+    public static final String LOG4J = "l";
+    public static final String WATCH = "w";
+    public static final String CONFIG = "c";
+    public static final String PROTOCOL = "protocol";
+
+    private final Set<Integer> _ports = new HashSet<Integer>();
+    private final Set<Integer> _sslPorts = new HashSet<Integer>();
+    private final Map<ProtocolExclusion,Set<Integer>> _exclusionMap = new HashMap<ProtocolExclusion, Set<Integer>>();
+
+    private String _configFile;
+    private String _logConfigFile;
+    private String _bind;
+    private String _transport = Transport.TCP;
+    private Integer _jmxPort;
+
+    private Integer _logWatchFrequency = 0;
+
+    public void addPort(final int port)
+    {
+        _ports.add(port);
+    }
+
+    public void addSSLPort(final int sslPort)
+    {
+        _sslPorts.add(sslPort);
+    }
+
+    public Set<Integer> getPorts()
+    {
+        return Collections.unmodifiableSet(_ports);
+    }
+
+    public Set<Integer> getSSLPorts()
+    {
+        return Collections.unmodifiableSet(_sslPorts);
+    }
+
+    public String getConfigFile()
+    {
+        return _configFile;
+    }
+
+    public void setConfigFile(final String configFile)
+    {
+        _configFile = configFile;
+    }
+
+    public String getLogConfigFile()
+    {
+        return _logConfigFile;
+    }
+
+    public void setLogConfigFile(final String logConfigFile)
+    {
+        _logConfigFile = logConfigFile;
+    }
+
+    public Integer getJmxPort()
+    {
+        return _jmxPort;
+    }
+
+    public void setJmxPort(final int jmxPort)
+    {
+        _jmxPort = jmxPort;
+    }
+
+    public String getQpidHome()
+    {
+        return System.getProperty(QPID_HOME);
+    }
+
+    public Set<Integer> getExcludedPorts(final ProtocolExclusion excludeProtocol)
+    {
+        final Set<Integer> excludedPorts = _exclusionMap.get(excludeProtocol);
+        return excludedPorts == null ? Collections.<Integer>emptySet() : excludedPorts;
+    }
+
+    public void addExcludedPort(final ProtocolExclusion excludeProtocol, final int port)
+    {
+        if (!_exclusionMap.containsKey(excludeProtocol))
+        {
+            _exclusionMap.put(excludeProtocol, new HashSet<Integer>());
+        }
+
+        Set<Integer> ports = _exclusionMap.get(excludeProtocol);
+        ports.add(port);
+    }
+
+    public String getBind()
+    {
+        return _bind;
+    }
+
+    public void setBind(final String bind)
+    {
+        _bind = bind;
+    }
+
+    public int getLogWatchFrequency()
+    {
+        return _logWatchFrequency;
+    }
+
+    public void setLogWatchFrequency(final int logWatchFrequency)
+    {
+        _logWatchFrequency = logWatchFrequency;
+    }
+
+    public String getTransport()
+    {
+        return _transport;
+    }
+
+    public void setTransport(final String transport)
+    {
+        _transport = transport;
+    }
+}
\ No newline at end of file
diff --git a/java/broker/src/main/java/org/apache/qpid/server/Main.java b/java/broker/src/main/java/org/apache/qpid/server/Main.java
index 2925db69de..317459942a 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/Main.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/Main.java
@@ -20,19 +20,6 @@
  */
 package org.apache.qpid.server;
 
-import static org.apache.qpid.transport.ConnectionSettings.WILDCARD_ADDRESS;
-
-import java.io.File;
-import java.io.IOException;
-import java.io.InputStream;
-import java.net.InetAddress;
-import java.net.InetSocketAddress;
-import java.util.EnumSet;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Properties;
-import java.util.Set;
-
 import org.apache.commons.cli.CommandLine;
 import org.apache.commons.cli.HelpFormatter;
 import org.apache.commons.cli.Option;
@@ -41,32 +28,9 @@ import org.apache.commons.cli.Options;
 import org.apache.commons.cli.ParseException;
 import org.apache.commons.cli.PosixParser;
 import org.apache.log4j.Logger;
-import org.apache.log4j.PropertyConfigurator;
-import org.apache.log4j.xml.QpidLog4JConfigurator;
-import org.apache.qpid.common.QpidProperties;
-import org.apache.qpid.framing.ProtocolVersion;
-import org.apache.qpid.server.configuration.ServerConfiguration;
-import org.apache.qpid.server.configuration.ServerNetworkTransportConfiguration;
-import org.apache.qpid.server.configuration.management.ConfigurationManagementMBean;
-import org.apache.qpid.server.information.management.ServerInformationMBean;
-import org.apache.qpid.server.logging.SystemOutMessageLogger;
-import org.apache.qpid.server.logging.actors.BrokerActor;
-import org.apache.qpid.server.logging.actors.CurrentActor;
-import org.apache.qpid.server.logging.actors.GenericActor;
-import org.apache.qpid.server.logging.management.LoggingManagementMBean;
-import org.apache.qpid.server.logging.messages.BrokerMessages;
-import org.apache.qpid.server.protocol.AMQProtocolEngineFactory;
-import org.apache.qpid.server.protocol.MultiVersionProtocolEngineFactory;
-import org.apache.qpid.server.protocol.MultiVersionProtocolEngineFactory.VERSION;
+import org.apache.qpid.server.Broker.InitException;
 import org.apache.qpid.server.registry.ApplicationRegistry;
-import org.apache.qpid.server.registry.ConfigurationFileApplicationRegistry;
-import org.apache.qpid.server.transport.QpidAcceptor;
-import org.apache.qpid.ssl.SSLContextFactory;
-import org.apache.qpid.transport.ConnectionSettings;
-import org.apache.qpid.transport.NetworkTransportConfiguration;
-import org.apache.qpid.transport.network.IncomingNetworkTransport;
-import org.apache.qpid.transport.network.Transport;
-import org.apache.qpid.transport.network.mina.MinaNetworkTransport;
+
 
 /**
  * Main entry point for AMQPD.
@@ -76,35 +40,38 @@ public class Main
 {
     private static Logger _logger;
 
-    private static final String DEFAULT_CONFIG_FILE = "etc/config.xml";
-
-    public static final String DEFAULT_LOG_CONFIG_FILENAME = "log4j.xml";
-    public static final String QPID_HOME = "QPID_HOME";
-    private static final int IPV4_ADDRESS_LENGTH = 4;
-
-    private static final char IPV4_LITERAL_SEPARATOR = '.';
+    protected final static Options options = new Options();
+    protected static CommandLine commandLine;
 
-    protected static class InitException extends Exception
+    public static void main(String[] args)
     {
-        InitException(String msg, Throwable cause)
+        //if the -Dlog4j.configuration property has not been set, enable the init override
+        //to stop Log4J wondering off and picking up the first log4j.xml/properties file it
+        //finds from the classpath when we get the first Loggers
+        if(System.getProperty("log4j.configuration") == null)
         {
-            super(msg, cause);
+            System.setProperty("log4j.defaultInitOverride", "true");
         }
-    }
 
-    protected final Options options = new Options();
-    protected CommandLine commandLine;
-
-    protected Main(String[] args)
-    {
+        //now that the override status is know, we can instantiate the Loggers
+        _logger = Logger.getLogger(Main.class);
         setOptions(options);
         if (parseCommandline(args))
         {
-            execute();
+            try
+            {
+                execute();
+            }
+            catch(Exception e)
+            {
+                System.err.println("Exception during startup: " + e);
+                e.printStackTrace();
+                shutdown(1);
+            }
         }
     }
 
-    protected boolean parseCommandline(String[] args)
+    protected static boolean parseCommandline(String[] args)
     {
         try
         {
@@ -122,8 +89,7 @@ public class Main
         }
     }
 
-    @SuppressWarnings("static-access")
-    protected void setOptions(Options options)
+    protected static void setOptions(Options options)
     {
         Option help = new Option("h", "help", false, "print this message");
         Option version = new Option("v", "version", false, "print the version information and exit");
@@ -171,7 +137,7 @@ public class Main
         Option logconfig =
                 OptionBuilder.withArgName("logconfig").hasArg()
                         .withDescription("use the specified log4j xml configuration file. By "
-                                         + "default looks for a file named " + DEFAULT_LOG_CONFIG_FILENAME
+                                         + "default looks for a file named " + BrokerOptions.DEFAULT_LOG_CONFIG_FILE
                                          + " in the same directory as the configuration file").withLongOpt("logconfig").create("l");
         Option logwatchconfig =
                 OptionBuilder.withArgName("logwatch").hasArg()
@@ -198,445 +164,110 @@ public class Main
         options.addOption(sslport);
     }
 
-    protected void execute()
+    protected static void execute() throws Exception
     {
-        // note this understands either --help or -h. If an option only has a long name you can use that but if
-        // an option has a short name and a long name you must use the short name here.
-        if (commandLine.hasOption("h"))
+        BrokerOptions options = new BrokerOptions();
+        String configFile = commandLine.getOptionValue(BrokerOptions.CONFIG);
+        if(configFile != null)
         {
-            HelpFormatter formatter = new HelpFormatter();
-            formatter.printHelp("Qpid", options, true);
+            options.setConfigFile(configFile);
         }
-        else if (commandLine.hasOption("v"))
-        {
-            String ver = QpidProperties.getVersionString();
 
-            StringBuilder protocol = new StringBuilder("AMQP version(s) [major.minor]: ");
-
-            boolean first = true;
-            for (ProtocolVersion pv : ProtocolVersion.getSupportedProtocolVersions())
-            {
-                if (first)
-                {
-                    first = false;
-                }
-                else
-                {
-                    protocol.append(", ");
-                }
-
-                protocol.append(pv.getMajorVersion()).append('-').append(pv.getMinorVersion());
-
-            }
-
-            System.out.println(ver + " (" + protocol + ")");
-        }
-        else
+        String logWatchConfig = commandLine.getOptionValue(BrokerOptions.WATCH);
+        if(logWatchConfig != null)
         {
-            try
-            {
-                CurrentActor.set(new BrokerActor(new SystemOutMessageLogger()));
-                startup();
-                CurrentActor.remove();
-            }
-            catch (InitException e)
-            {
-                System.out.println("Initialisation Error : " + e.getMessage());
-                shutdown(1);
-            }
-            catch (Throwable e)
-            {
-                System.out.println("Error initialising message broker: " + e);
-                e.printStackTrace();
-                shutdown(1);
-            }
+            options.setLogWatchFrequency(Integer.parseInt(logWatchConfig) * 1000);
         }
-    }
 
-    protected void shutdown(int status)
-    {
-        ApplicationRegistry.remove();
-        System.exit(status);
-    }
-
-    protected void startup() throws Exception
-    {
-        final String QpidHome = System.getProperty(QPID_HOME);
-        final File defaultConfigFile = new File(QpidHome, DEFAULT_CONFIG_FILE);
-        final File configFile = new File(commandLine.getOptionValue("c", defaultConfigFile.getPath()));
-        if (!configFile.exists())
+        String logConfig = commandLine.getOptionValue(BrokerOptions.LOG4J);
+        if(logConfig != null)
         {
-            String error = "File " + configFile + " could not be found. Check the file exists and is readable.";
-
-            if (QpidHome == null)
-            {
-                error = error + "\nNote: " + QPID_HOME + " is not set.";
-            }
-
-            throw new InitException(error, null);
-        }
-        else
-        {
-            CurrentActor.get().message(BrokerMessages.CONFIG(configFile.getAbsolutePath()));
+            options.setLogConfigFile(logConfig);
         }
 
-        String logConfig = commandLine.getOptionValue("l");
-        String logWatchConfig = commandLine.getOptionValue("w", "0");
-
-        int logWatchTime = 0;
-        try
-        {
-            logWatchTime = Integer.parseInt(logWatchConfig);
-        }
-        catch (NumberFormatException e)
+        String jmxPort = commandLine.getOptionValue(BrokerOptions.MANAGEMENT);
+        if(jmxPort != null)
         {
-            System.err.println("Log watch configuration value of " + logWatchConfig + " is invalid. Must be "
-                               + "a non-negative integer. Using default of zero (no watching configured");
+            options.setJmxPort(Integer.parseInt(jmxPort));
         }
 
-        File logConfigFile;
-        if (logConfig != null)
-        {
-            logConfigFile = new File(logConfig);
-            configureLogging(logConfigFile, logWatchTime);
-        }
-        else
+        String bindAddr = commandLine.getOptionValue(BrokerOptions.BIND);
+        if (bindAddr != null)
         {
-            File configFileDirectory = configFile.getParentFile();
-            logConfigFile = new File(configFileDirectory, DEFAULT_LOG_CONFIG_FILENAME);
-            configureLogging(logConfigFile, logWatchTime);
+            options.setBind(bindAddr);
         }
 
-        ConfigurationFileApplicationRegistry config = new ConfigurationFileApplicationRegistry(configFile);
-        ServerConfiguration serverConfig = config.getConfiguration();
-        updateManagementPort(serverConfig, commandLine.getOptionValue("m"));
-
-        ApplicationRegistry.initialise(config);
-
-        // We have already loaded the BrokerMessages class by this point so we
-        // need to refresh the locale setting incase we had a different value in
-        // the configuration.
-        BrokerMessages.reload();
-
-        // AR.initialise() sets and removes its own actor so we now need to set the actor
-        // for the remainder of the startup, and the default actor if the stack is empty
-        CurrentActor.set(new BrokerActor(config.getCompositeStartupMessageLogger()));
-        CurrentActor.setDefault(new BrokerActor(config.getRootMessageLogger()));
-        GenericActor.setDefaultMessageLogger(config.getRootMessageLogger());
-        
-
-        try
+        String[] portStr = commandLine.getOptionValues(BrokerOptions.PORTS);
+        if(portStr != null)
         {
-            configureLoggingManagementMBean(logConfigFile, logWatchTime);
-
-            ConfigurationManagementMBean configMBean = new ConfigurationManagementMBean();
-            configMBean.register();
-
-            ServerInformationMBean sysInfoMBean = new ServerInformationMBean(config);
-            sysInfoMBean.register();
-
-
-            String[] portStr = commandLine.getOptionValues("p");
-
-            Set<Integer> ports = new HashSet<Integer>();
-            Set<Integer> exclude_0_10 = new HashSet<Integer>();
-            Set<Integer> exclude_0_9_1 = new HashSet<Integer>();
-            Set<Integer> exclude_0_9 = new HashSet<Integer>();
-            Set<Integer> exclude_0_8 = new HashSet<Integer>();
-
-            if(portStr == null || portStr.length == 0)
-            {
-
-                parsePortList(ports, serverConfig.getPorts());
-                parsePortList(exclude_0_10, serverConfig.getPortExclude010());
-                parsePortList(exclude_0_9_1, serverConfig.getPortExclude091());
-                parsePortList(exclude_0_9, serverConfig.getPortExclude09());
-                parsePortList(exclude_0_8, serverConfig.getPortExclude08());
-
-            }
-            else
+            parsePortArray(options, portStr, false);
+            for(ProtocolExclusion pe : ProtocolExclusion.values())
             {
-                parsePortArray(ports, portStr);
-                parsePortArray(exclude_0_10, commandLine.getOptionValues("exclude-0-10"));
-                parsePortArray(exclude_0_9_1, commandLine.getOptionValues("exclude-0-9-1"));
-                parsePortArray(exclude_0_9, commandLine.getOptionValues("exclude-0-9"));
-                parsePortArray(exclude_0_8, commandLine.getOptionValues("exclude-0-8"));
-
-            }
-
-
-
-
-            String bindAddr = commandLine.getOptionValue("b");
-            if (bindAddr == null)
-            {
-                bindAddr = serverConfig.getBind();
-            }
-            InetAddress bindAddress = null;
-
-
-
-            if (bindAddr.equals(WILDCARD_ADDRESS))
-            {
-                bindAddress = new InetSocketAddress(0).getAddress();
-            }
-            else
-            {
-                bindAddress = InetAddress.getByAddress(parseIP(bindAddr));
-            }
-
-            String hostName = bindAddress.getCanonicalHostName();
-
-
-            String keystorePath = serverConfig.getKeystorePath();
-            String keystorePassword = serverConfig.getKeystorePassword();
-            String certType = serverConfig.getCertType();
-            SSLContextFactory sslFactory = null;
-            
-            if (!serverConfig.getSSLOnly())
-            {
-
-                for(int port : ports)
-                {
-                    Set<VERSION> supported = EnumSet.allOf(VERSION.class);
-
-                    if(exclude_0_10.contains(port))
-                    {
-                        supported.remove(VERSION.v0_10);
-                    }
-
-                    if(exclude_0_9_1.contains(port))
-                    {
-                        supported.remove(VERSION.v0_9_1);
-                    }
-                    if(exclude_0_9.contains(port))
-                    {
-                        supported.remove(VERSION.v0_9);
-                    }
-                    if(exclude_0_8.contains(port))
-                    {
-                        supported.remove(VERSION.v0_8);
-                    }
-
-                    NetworkTransportConfiguration settings = 
-                        new ServerNetworkTransportConfiguration(serverConfig, port, bindAddress.getHostName(), Transport.TCP);
-
-                    IncomingNetworkTransport transport = new MinaNetworkTransport();
-                    MultiVersionProtocolEngineFactory protocolEngineFactory =
-                            new MultiVersionProtocolEngineFactory(hostName, supported);
-
-                    transport.accept(settings, protocolEngineFactory, sslFactory);
-                    ApplicationRegistry.getInstance().addAcceptor(new InetSocketAddress(bindAddress, port),
-                                                                  new QpidAcceptor(transport, Transport.TCP));
-                    CurrentActor.get().message(BrokerMessages.LISTENING("TCP", port));
-                }
-
+                parsePortArray(options, commandLine.getOptionValues(pe.getExcludeName()), pe);
             }
+        }
 
-            if (serverConfig.getEnableSSL())
+        String[] sslPortStr = commandLine.getOptionValues(BrokerOptions.SSL_PORTS);
+        if(sslPortStr != null)
+        {
+            parsePortArray(options, sslPortStr, true);
+            for(ProtocolExclusion pe : ProtocolExclusion.values())
             {
-                String sslPort = commandLine.getOptionValue("s");
-                int port = 0;
-                if (null != sslPort)
-                {
-                    port = Integer.parseInt(sslPort);
-                }
-                else
-                {
-                    port = serverConfig.getSSLPort();
-                }
-
-                NetworkTransportConfiguration settings = 
-                    new ServerNetworkTransportConfiguration(serverConfig, port, bindAddress.getHostName(), Transport.TCP);
-
-                sslFactory = new SSLContextFactory(keystorePath, keystorePassword, certType);
-
-                IncomingNetworkTransport transport = new MinaNetworkTransport();
-
-                transport.accept(settings, new AMQProtocolEngineFactory(), sslFactory);
-
-                ApplicationRegistry.getInstance().addAcceptor(new InetSocketAddress(bindAddress, port),
-                                                              new QpidAcceptor(transport,"TCP"));
-                CurrentActor.get().message(BrokerMessages.LISTENING("TCP/SSL", port));
+                parsePortArray(options, commandLine.getOptionValues(pe.getExcludeName()), pe);
             }
-
-            CurrentActor.get().message(BrokerMessages.READY());
-
-        }
-        finally
-        {
-            // Startup is complete so remove the AR initialised Startup actor
-            CurrentActor.remove();
         }
-
-
-
+        
+        Broker broker = new Broker();
+        broker.startup(options);
     }
 
-    private void parsePortArray(Set<Integer> ports, String[] portStr)
-            throws InitException
+    protected static void shutdown(int status)
     {
-        if(portStr != null)
-        {
-            for(int i = 0; i < portStr.length; i++)
-            {
-                try
-                {
-                    ports.add(Integer.parseInt(portStr[i]));
-                }
-                catch (NumberFormatException e)
-                {
-                    throw new InitException("Invalid port: " + portStr[i], e);
-                }
-            }
-        }
+        ApplicationRegistry.remove();
+        System.exit(status);
     }
 
-    private void parsePortList(Set<Integer> output, List input)
-            throws InitException
+    private static void parsePortArray(BrokerOptions options, Object[] ports, boolean ssl) throws InitException
     {
-        if(input != null)
+        if(ports != null)
         {
-            for(Object port : input)
+            for(int i = 0; i < ports.length; i++)
             {
                 try
                 {
-                    output.add(Integer.parseInt(String.valueOf(port)));
+                    if(ssl)
+                    {
+                        options.addSSLPort(Integer.parseInt(String.valueOf(ports[i])));
+                    }
+                    else
+                    {
+                        options.addPort(Integer.parseInt(String.valueOf(ports[i])));
+                    }
                 }
                 catch (NumberFormatException e)
                 {
-                    throw new InitException("Invalid port: " + port, e);
+                    throw new InitException("Invalid port: " + ports[i], e);
                 }
             }
         }
     }
 
-    /**
-     * Update the configuration data with the management port.
-     * @param configuration
-     * @param managementPort The string from the command line
-     */
-    private void updateManagementPort(ServerConfiguration configuration, String managementPort)
-    {
-        if (managementPort != null)
-        {
-            try
-            {
-                configuration.setJMXManagementPort(Integer.parseInt(managementPort));
-            }
-            catch (NumberFormatException e)
-            {
-                _logger.warn("Invalid management port: " + managementPort + " will use:" + configuration.getJMXManagementPort(), e);
-            }
-        }
-    }
-
-    public static void main(String[] args)
-    {
-        //if the -Dlog4j.configuration property has not been set, enable the init override
-        //to stop Log4J wondering off and picking up the first log4j.xml/properties file it
-        //finds from the classpath when we get the first Loggers
-        if(System.getProperty("log4j.configuration") == null)
-        {
-            System.setProperty("log4j.defaultInitOverride", "true");
-        }
-
-        //now that the override status is know, we can instantiate the Loggers
-        _logger = Logger.getLogger(Main.class);
-
-        new Main(args);
-    }
-
-    private byte[] parseIP(String address) throws Exception
-    {
-        char[] literalBuffer = address.toCharArray();
-        int byteCount = 0;
-        int currByte = 0;
-        byte[] ip = new byte[IPV4_ADDRESS_LENGTH];
-        for (int i = 0; i < literalBuffer.length; i++)
-        {
-            char currChar = literalBuffer[i];
-            if ((currChar >= '0') && (currChar <= '9'))
-            {
-                currByte = (currByte * 10) + (Character.digit(currChar, 10) & 0xFF);
-            }
-
-            if (currChar == IPV4_LITERAL_SEPARATOR || (i + 1 == literalBuffer.length))
-            {
-                ip[byteCount++] = (byte) currByte;
-                currByte = 0;
-            }
-        }
-
-        if (byteCount != 4)
-        {
-            throw new Exception("Invalid IP address: " + address);
-        }
-        return ip;
-    }
-
-    private void configureLogging(File logConfigFile, int logWatchTime) throws InitException, IOException
+    private static void parsePortArray(BrokerOptions options, Object[] ports, ProtocolExclusion excludedProtocol) throws InitException
     {
-        if (logConfigFile.exists() && logConfigFile.canRead())
+        if(ports != null)
         {
-            CurrentActor.get().message(BrokerMessages.LOG_CONFIG(logConfigFile.getAbsolutePath()));
-
-            if (logWatchTime > 0)
-            {
-                System.out.println("log file " + logConfigFile.getAbsolutePath() + " will be checked for changes every "
-                                   + logWatchTime + " seconds");
-                // log4j expects the watch interval in milliseconds
-                try
-                {
-                    QpidLog4JConfigurator.configureAndWatch(logConfigFile.getPath(), logWatchTime * 1000);
-                }
-                catch (Exception e)
-                {
-                    throw new InitException(e.getMessage(),e);
-                }
-            }
-            else
+            for(int i = 0; i < ports.length; i++)
             {
                 try
                 {
-                    QpidLog4JConfigurator.configure(logConfigFile.getPath());
+                    options.addExcludedPort(excludedProtocol, 
+                            Integer.parseInt(String.valueOf(ports[i])));
                 }
-                catch (Exception e)
-                {
-                    throw new InitException(e.getMessage(),e);
-                }
-            }
-        }
-        else
-        {
-            System.err.println("Logging configuration error: unable to read file " + logConfigFile.getAbsolutePath());
-            System.err.println("Using the fallback internal log4j.properties configuration");
-
-            InputStream propsFile = this.getClass().getResourceAsStream("/log4j.properties");
-            if(propsFile == null)
-            {
-                throw new IOException("Unable to load the fallback internal log4j.properties configuration file");
-            }
-            else
-            {
-                try
-                {
-                    Properties fallbackProps = new Properties();
-                    fallbackProps.load(propsFile);
-                    PropertyConfigurator.configure(fallbackProps);
-                }
-                finally
+                catch (NumberFormatException e)
                 {
-                    propsFile.close();
+                    throw new InitException("Invalid port for exclusion: " + ports[i], e);
                 }
             }
         }
     }
-
-    private void configureLoggingManagementMBean(File logConfigFile, int logWatchTime) throws Exception
-    {
-        LoggingManagementMBean blm = new LoggingManagementMBean(logConfigFile.getPath(),logWatchTime);
-
-        blm.register();
-    }
 }
diff --git a/java/broker/src/main/java/org/apache/qpid/server/ProtocolExclusion.java b/java/broker/src/main/java/org/apache/qpid/server/ProtocolExclusion.java
new file mode 100644
index 0000000000..22d97d36dd
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/ProtocolExclusion.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;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public enum ProtocolExclusion
+{
+    v0_8("exclude-0-8","--exclude-0-8"),
+    v0_9("exclude-0-9", "--exclude-0-9"),
+    v0_9_1("exclude-0-9-1", "--exclude-0-9-1"),
+    v0_10("exclude-0-10", "--exclude-0-10");
+
+    private static final Map<String, ProtocolExclusion> MAP = new HashMap<String,ProtocolExclusion>();
+
+    static
+    {
+        for(ProtocolExclusion pe : ProtocolExclusion.values())
+        {
+            MAP.put(pe.getArg(), pe);
+        }
+    }
+
+    private String _arg;
+    private String _excludeName;
+
+    private ProtocolExclusion(final String excludeName, final String arg)
+    {
+        _excludeName = excludeName;
+        _arg = arg;
+    }
+
+    public String getArg()
+    {
+        return _arg;
+    }
+
+    public String getExcludeName()
+    {
+        return _excludeName;
+    }
+
+    public static ProtocolExclusion lookup(final String arg)
+    {
+        ProtocolExclusion ex = MAP.get(arg);
+
+        if(ex == null)
+        {
+            throw new IllegalArgumentException(arg + " is not a valid protocol exclusion");
+        }
+
+        return ex;
+    }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java b/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java
index 23ab5e8222..14de7c1723 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java
@@ -638,7 +638,7 @@ public class ServerConfiguration extends ConfigurationPlugin implements SignalHa
 
     public List getPorts()
     {
-        return getListValue("connector.port", Collections.singletonList(DEFAULT_PORT));
+        return getListValue("connector.port", Collections.<Integer>singletonList(DEFAULT_PORT));
     }
 
     public List getPortExclude010()
@@ -696,9 +696,9 @@ public class ServerConfiguration extends ConfigurationPlugin implements SignalHa
         return getBooleanValue("connector.ssl.sslOnly");
     }
 
-    public int getSSLPort()
+    public List getSSLPorts()
     {
-        return getIntValue("connector.ssl.port", DEFAULT_SSL_PORT);
+        return getListValue("connector.ssl.port", Collections.<Integer>singletonList(DEFAULT_SSL_PORT));
     }
 
     public String getKeystorePath()
diff --git a/java/broker/src/test/java/org/apache/qpid/server/BrokerOptionsTest.java b/java/broker/src/test/java/org/apache/qpid/server/BrokerOptionsTest.java
new file mode 100644
index 0000000000..35ab28656b
--- /dev/null
+++ b/java/broker/src/test/java/org/apache/qpid/server/BrokerOptionsTest.java
@@ -0,0 +1,210 @@
+/*
+ *
+ * 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;
+
+import static org.apache.qpid.transport.ConnectionSettings.WILDCARD_ADDRESS;
+import static org.apache.qpid.transport.network.Transport.TCP;
+import static org.apache.qpid.transport.network.Transport.VM;
+import static org.apache.qpid.server.configuration.ServerConfiguration.DEFAULT_PORT;
+import static org.apache.qpid.server.configuration.ServerConfiguration.DEFAULT_JMXPORT;
+
+import java.util.Collections;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Set;
+
+
+import org.apache.qpid.test.utils.QpidTestCase;
+
+
+public class BrokerOptionsTest extends QpidTestCase
+{
+    private BrokerOptions _options;
+    
+    private static final int TEST_PORT1 = 6789;
+    private static final int TEST_PORT2 = 6790;
+    
+
+    protected void setUp()
+    {
+        _options = new BrokerOptions();
+    }
+    
+    public void testDefaultPort()
+    {
+        assertEquals(Collections.<Integer>emptySet(), _options.getPorts());
+    }
+
+    public void testOverriddenPort()
+    {
+        _options.addPort(TEST_PORT1);
+        assertEquals(Collections.singleton(TEST_PORT1), _options.getPorts());
+    }
+
+    public void testManyOverriddenPorts()
+    {
+        _options.addPort(TEST_PORT1);
+        _options.addPort(TEST_PORT2);
+        final Set<Integer> expectedPorts = new HashSet<Integer>(Arrays.asList(new Integer[] {TEST_PORT1, TEST_PORT2}));
+        assertEquals(expectedPorts, _options.getPorts());
+    }
+
+    public void testDuplicateOverriddenPortsAreSilentlyIgnored()
+    {
+        _options.addPort(TEST_PORT1);
+        _options.addPort(TEST_PORT2);
+        _options.addPort(TEST_PORT1); // duplicate - should be silently ignored
+        final Set<Integer> expectedPorts = new HashSet<Integer>(Arrays.asList(new Integer[] {TEST_PORT1, TEST_PORT2}));
+        assertEquals(expectedPorts, _options.getPorts());
+    }
+
+    public void testDefaultSSLPort()
+    {
+        assertEquals(Collections.<Integer>emptySet(), _options.getSSLPorts());
+    }
+
+    public void testOverriddenSSLPort()
+    {
+        _options.addSSLPort(TEST_PORT1);
+        assertEquals(Collections.singleton(TEST_PORT1), _options.getSSLPorts());
+    }
+
+    public void testManyOverriddenSSLPorts()
+    {
+        _options.addSSLPort(TEST_PORT1);
+        _options.addSSLPort(TEST_PORT2);
+        final Set<Integer> expectedPorts = new HashSet<Integer>(Arrays.asList(new Integer[] {TEST_PORT1, TEST_PORT2}));
+        assertEquals(expectedPorts, _options.getSSLPorts());
+    }
+
+    public void testDuplicateOverriddenSSLPortsAreSilentlyIgnored()
+    {
+        _options.addSSLPort(TEST_PORT1);
+        _options.addSSLPort(TEST_PORT2);
+        _options.addSSLPort(TEST_PORT1); // duplicate - should be silently ignored
+        final Set<Integer> expectedPorts = new HashSet<Integer>(Arrays.asList(new Integer[] {TEST_PORT1, TEST_PORT2}));
+        assertEquals(expectedPorts, _options.getSSLPorts());
+    }
+
+    public void testDefaultConfigFile()
+    {
+        assertNull(_options.getConfigFile());
+    }
+    
+    public void testOverriddenConfigFile()
+    {
+        final String testConfigFile = "etc/mytestconfig.xml";
+        _options.setConfigFile(testConfigFile);
+        assertEquals(testConfigFile, _options.getConfigFile());
+    }
+
+    public void testDefaultLogConfigFile()
+    {
+        assertNull(_options.getLogConfigFile());
+    }
+
+    public void testOverriddenLogConfigFile()
+    {
+        final String testLogConfigFile = "etc/mytestlog4j.xml";
+        _options.setLogConfigFile(testLogConfigFile);
+        assertEquals(testLogConfigFile, _options.getLogConfigFile());
+    }
+
+    public void testDefaultJmxPort()
+    {
+        assertNull(_options.getJmxPort());
+    }
+
+    public void testJmxPort()
+    {
+        _options.setJmxPort(TEST_PORT1);
+        assertEquals(Integer.valueOf(TEST_PORT1), _options.getJmxPort());
+    }
+
+    public void testQpidHomeExposesSysProperty()
+    {
+        assertEquals(System.getProperty("QPID_HOME"), _options.getQpidHome());
+    }
+    
+    public void testDefaultExcludesPortFor0_10()
+    {
+        assertEquals(Collections.EMPTY_SET, _options.getExcludedPorts(ProtocolExclusion.v0_10));
+    }
+    
+    public void testOverriddenExcludesPortFor0_10()
+    {
+        _options.addExcludedPort(ProtocolExclusion.v0_10, TEST_PORT1);
+        assertEquals(Collections.singleton(TEST_PORT1), _options.getExcludedPorts(ProtocolExclusion.v0_10));
+    }
+
+    public void testManyOverriddenExcludedPortFor0_10()
+    {
+        _options.addExcludedPort(ProtocolExclusion.v0_10, TEST_PORT1);
+        _options.addExcludedPort(ProtocolExclusion.v0_10, TEST_PORT2);
+        final Set<Integer> expectedPorts = new HashSet<Integer>(Arrays.asList(new Integer[] {TEST_PORT1, TEST_PORT2}));
+        assertEquals(expectedPorts, _options.getExcludedPorts(ProtocolExclusion.v0_10));
+    }
+
+    public void testDuplicatedOverriddenExcludedPortFor0_10AreSilentlyIgnored()
+    {
+        _options.addExcludedPort(ProtocolExclusion.v0_10, TEST_PORT1);
+        _options.addExcludedPort(ProtocolExclusion.v0_10, TEST_PORT2);
+        final Set<Integer> expectedPorts = new HashSet<Integer>(Arrays.asList(new Integer[] {TEST_PORT1, TEST_PORT2}));
+        assertEquals(expectedPorts, _options.getExcludedPorts(ProtocolExclusion.v0_10));
+    }
+    
+    public void testDefaultBind()
+    {
+        assertNull(_options.getBind());
+    }
+    
+    public void testOverriddenBind()
+    {
+        final String bind = "192.168.0.1";
+        _options.setBind(bind);
+        assertEquals(bind, _options.getBind());
+    }
+
+    public void testDefaultLogWatchFrequency()
+    {
+        assertEquals(0L, _options.getLogWatchFrequency());
+    }
+
+    public void testOverridenLogWatchFrequency()
+    {
+        final int myFreq = 10 * 1000;
+        
+        _options.setLogWatchFrequency(myFreq);
+        assertEquals(myFreq, _options.getLogWatchFrequency());
+    }
+    
+    public void testDefaultTransport()
+    {
+        assertEquals(TCP, _options.getTransport());
+    }
+    
+    public void testOverriddenTransport()
+    {
+        _options.setTransport(VM);
+        
+        assertEquals(VM, _options.getTransport());
+    }
+}
diff --git a/java/broker/src/test/java/org/apache/qpid/server/RunBrokerWithCommand.java b/java/broker/src/test/java/org/apache/qpid/server/RunBrokerWithCommand.java
deleted file mode 100644
index 59543874b4..0000000000
--- a/java/broker/src/test/java/org/apache/qpid/server/RunBrokerWithCommand.java
+++ /dev/null
@@ -1,132 +0,0 @@
-/*
- *  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;
-
-import org.apache.log4j.Logger;
-import org.apache.log4j.Level;
-
-import java.io.InputStream;
-import java.io.BufferedReader;
-import java.io.InputStreamReader;
-import java.io.IOException;
-
-public class RunBrokerWithCommand
-{
-    public static void main(String[] args)
-    {
-        //Start the broker
-        try
-        {
-            String[] fudge = args.clone();
-
-            // Override the first value which is the command we are going to run later.
-            fudge[0] = "-v";
-            new Main(fudge).startup();
-        }
-        catch (Exception e)
-        {
-            System.err.println("Unable to start broker due to: " + e.getMessage());
-
-            e.printStackTrace();
-            exit(1);
-        }
-
-        Logger.getRootLogger().setLevel(Level.ERROR);
-
-        //run command
-        try
-        {
-            Process task = Runtime.getRuntime().exec(args[0]);
-            System.err.println("Started Proccess: " + args[0]);
-
-            InputStream inputStream = task.getInputStream();
-
-            InputStream errorStream = task.getErrorStream();
-
-            Thread out = new Thread(new Outputter("[OUT]", new BufferedReader(new InputStreamReader(inputStream))));
-            Thread err = new Thread(new Outputter("[ERR]", new BufferedReader(new InputStreamReader(errorStream))));
-
-            out.start();
-            err.start();
-
-            out.join();
-            err.join();
-
-            System.err.println("Waiting for process to exit: " + args[0]);
-            task.waitFor();
-            System.err.println("Done Proccess: " + args[0]);
-
-        }
-        catch (IOException e)
-        {
-            System.err.println("Proccess had problems: " + e.getMessage());
-            e.printStackTrace(System.err);
-            exit(1);
-        }
-        catch (InterruptedException e)
-        {
-            System.err.println("Proccess had problems: " + e.getMessage());
-            e.printStackTrace(System.err);
-
-            exit(1);
-        }
-
-
-        exit(0);
-    }
-
-    private static void exit(int i)
-    {
-        Logger.getRootLogger().setLevel(Level.INFO);
-        System.exit(i);
-    }
-
-    static class Outputter implements Runnable
-    {
-
-        BufferedReader reader;
-        String prefix;
-
-        Outputter(String s, BufferedReader r)
-        {
-            prefix = s;
-            reader = r;
-        }
-
-        public void run()
-        {
-            String line;
-            try
-            {
-                while ((line = reader.readLine()) != null)
-                {
-                    System.out.println(prefix + line);
-                }
-            }
-            catch (IOException e)
-            {
-                System.out.println("Error occured reading; " + e.getMessage());
-            }
-        }
-
-    }
-
-}
diff --git a/java/broker/src/test/java/org/apache/qpid/server/configuration/ServerConfigurationTest.java b/java/broker/src/test/java/org/apache/qpid/server/configuration/ServerConfigurationTest.java
index dcc8bda71a..c8a35e4405 100644
--- a/java/broker/src/test/java/org/apache/qpid/server/configuration/ServerConfigurationTest.java
+++ b/java/broker/src/test/java/org/apache/qpid/server/configuration/ServerConfigurationTest.java
@@ -59,8 +59,6 @@ public class ServerConfigurationTest extends QpidTestCase
         ApplicationRegistry.remove();
     }
 
-
-
     public void testSetJMXManagementPort() throws ConfigurationException
     {
         _serverConfig.initialise();
@@ -453,7 +451,7 @@ public class ServerConfigurationTest extends QpidTestCase
         assertEquals(10, _serverConfig.getConnectorProcessors());
         }
 
-    public void testGetPort() throws ConfigurationException
+    public void testGetPorts() throws ConfigurationException
     {
         // Check default
         _serverConfig.initialise();
@@ -562,17 +560,22 @@ public class ServerConfigurationTest extends QpidTestCase
         assertEquals(true, _serverConfig.getSSLOnly());
     }
 
-    public void testGetSSLPort() throws ConfigurationException
+    public void testGetSSLPorts() throws ConfigurationException
     {
         // Check default
         _serverConfig.initialise();
-        assertEquals(8672, _serverConfig.getSSLPort());
+        assertNotNull(_serverConfig.getSSLPorts());
+        assertEquals(1, _serverConfig.getSSLPorts().size());
+        assertEquals(ServerConfiguration.DEFAULT_SSL_PORT, _serverConfig.getSSLPorts().get(0));
+
 
         // Check value we set
-        _config.setProperty("connector.ssl.port", 23);
+        _config.setProperty("connector.ssl.port", "10");
         _serverConfig = new ServerConfiguration(_config);
         _serverConfig.initialise();
-        assertEquals(23, _serverConfig.getSSLPort());
+        assertNotNull(_serverConfig.getSSLPorts());
+        assertEquals(1, _serverConfig.getSSLPorts().size());
+        assertEquals("10", _serverConfig.getSSLPorts().get(0));
     }
 
     public void testGetKeystorePath() throws ConfigurationException
@@ -651,7 +654,7 @@ public class ServerConfigurationTest extends QpidTestCase
         out.close();
         ServerConfiguration conf = new ServerConfiguration(fileA);
         conf.initialise();
-        assertEquals(4235, conf.getSSLPort());
+        assertEquals("4235", conf.getSSLPorts().get(0));
     }
 
     public void testCombinedConfiguration() throws IOException, ConfigurationException
@@ -681,7 +684,7 @@ public class ServerConfigurationTest extends QpidTestCase
 
         ServerConfiguration config = new ServerConfiguration(mainFile.getAbsoluteFile());
         config.initialise();
-        assertEquals(4235, config.getSSLPort()); // From first file, not
+        assertEquals("4235", config.getSSLPorts().get(0)); // From first file, not
                                                  // overriden by second
         assertNotNull(config.getPorts());
         assertEquals(1, config.getPorts().size());
diff --git a/java/broker/src/test/java/org/apache/qpid/server/util/TestApplicationRegistry.java b/java/broker/src/test/java/org/apache/qpid/server/util/TestApplicationRegistry.java
index af8997cf40..31b4c20042 100644
--- a/java/broker/src/test/java/org/apache/qpid/server/util/TestApplicationRegistry.java
+++ b/java/broker/src/test/java/org/apache/qpid/server/util/TestApplicationRegistry.java
@@ -22,19 +22,29 @@ package org.apache.qpid.server.util;
 
 import org.apache.commons.configuration.ConfigurationException;
 import org.apache.qpid.server.configuration.ServerConfiguration;
+import org.apache.qpid.server.logging.NullRootMessageLogger;
+import org.apache.qpid.server.logging.actors.BrokerActor;
+import org.apache.qpid.server.logging.actors.CurrentActor;
 import org.apache.qpid.server.registry.ApplicationRegistry;
 import org.apache.qpid.server.security.auth.database.PropertiesPrincipalDatabaseManager;
 
 import java.util.Properties;
 
-
 public class TestApplicationRegistry extends ApplicationRegistry
 {
+
     public TestApplicationRegistry(ServerConfiguration config) throws ConfigurationException
     {
         super(config);
     }
 
+    @Override
+    public void initialise() throws Exception
+    {
+        CurrentActor.setDefault(new BrokerActor(new NullRootMessageLogger()));
+        super.initialise();
+    }
+
     protected void createDatabaseManager(ServerConfiguration configuration) throws Exception
     {
         Properties users = new Properties();
diff --git a/java/systests/src/main/java/org/apache/qpid/client/MultipleJCAProviderRegistrationTest.java b/java/systests/src/main/java/org/apache/qpid/client/MultipleJCAProviderRegistrationTest.java
index 29b4dd82a7..b50643583e 100644
--- a/java/systests/src/main/java/org/apache/qpid/client/MultipleJCAProviderRegistrationTest.java
+++ b/java/systests/src/main/java/org/apache/qpid/client/MultipleJCAProviderRegistrationTest.java
@@ -40,7 +40,7 @@ public class MultipleJCAProviderRegistrationTest extends QpidBrokerTestCase
 
     public void setUp() throws Exception
     {
-        _broker = VM;
+        _brokerType = VM;
 
         super.setUp();
     }
diff --git a/java/systests/src/main/java/org/apache/qpid/management/jmx/MessageStatisticsConfigurationTest.java b/java/systests/src/main/java/org/apache/qpid/management/jmx/MessageStatisticsConfigurationTest.java
index df8c6e74cd..7cc5d5995a 100644
--- a/java/systests/src/main/java/org/apache/qpid/management/jmx/MessageStatisticsConfigurationTest.java
+++ b/java/systests/src/main/java/org/apache/qpid/management/jmx/MessageStatisticsConfigurationTest.java
@@ -55,7 +55,7 @@ public class MessageStatisticsConfigurationTest extends MessageStatisticsTestCas
         assertEquals("Incorrect vhost data", 0, vhost.getTotalDataReceived());
         assertFalse("Vhost statistics should not be enabled", vhost.isStatisticsEnabled());
 
-        if (!_broker.equals(VM))
+        if (!_brokerType.equals(VM))
         {
             assertEquals("Incorrect server total messages", 5, _jmxUtils.getServerInformation().getTotalMessagesReceived());
             assertEquals("Incorrect server total data", 1000, _jmxUtils.getServerInformation().getTotalDataReceived());
@@ -83,7 +83,7 @@ public class MessageStatisticsConfigurationTest extends MessageStatisticsTestCas
         assertEquals("Incorrect vhost data", 1000, vhost.getTotalDataReceived());
         assertTrue("Vhost statistics should be enabled", vhost.isStatisticsEnabled());
 
-        if (!_broker.equals(VM))
+        if (!_brokerType.equals(VM))
         {
             assertEquals("Incorrect server total messages", 0, _jmxUtils.getServerInformation().getTotalMessagesReceived());
             assertEquals("Incorrect server total data", 0, _jmxUtils.getServerInformation().getTotalDataReceived());
@@ -111,7 +111,7 @@ public class MessageStatisticsConfigurationTest extends MessageStatisticsTestCas
         assertEquals("Incorrect vhost data", 0, vhost.getTotalDataReceived());
         assertFalse("Vhost statistics should not be enabled", vhost.isStatisticsEnabled());
 
-        if (!_broker.equals(VM))
+        if (!_brokerType.equals(VM))
         {
             assertEquals("Incorrect server total messages", 0, _jmxUtils.getServerInformation().getTotalMessagesReceived());
             assertEquals("Incorrect server total data", 0, _jmxUtils.getServerInformation().getTotalDataReceived());
@@ -139,7 +139,7 @@ public class MessageStatisticsConfigurationTest extends MessageStatisticsTestCas
         assertEquals("Incorrect vhost data", 1000, vhost.getTotalDataReceived());
         assertTrue("Vhost statistics should be enabled", vhost.isStatisticsEnabled());
 
-        if (!_broker.equals(VM))
+        if (!_brokerType.equals(VM))
         {
             assertEquals("Incorrect server total messages", 5, _jmxUtils.getServerInformation().getTotalMessagesReceived());
             assertEquals("Incorrect server total data", 1000, _jmxUtils.getServerInformation().getTotalDataReceived());
@@ -167,7 +167,7 @@ public class MessageStatisticsConfigurationTest extends MessageStatisticsTestCas
         assertEquals("Incorrect vhost data", 1000, vhost.getTotalDataReceived());
         assertTrue("Vhost statistics should be enabled", vhost.isStatisticsEnabled());
 
-        if (!_broker.equals(VM))
+        if (!_brokerType.equals(VM))
         {
             assertEquals("Incorrect server total messages", 5, _jmxUtils.getServerInformation().getTotalMessagesReceived());
             assertEquals("Incorrect server total data", 1000, _jmxUtils.getServerInformation().getTotalDataReceived());
diff --git a/java/systests/src/main/java/org/apache/qpid/management/jmx/MessageStatisticsTest.java b/java/systests/src/main/java/org/apache/qpid/management/jmx/MessageStatisticsTest.java
index 50ca51b18a..f1953d1eba 100644
--- a/java/systests/src/main/java/org/apache/qpid/management/jmx/MessageStatisticsTest.java
+++ b/java/systests/src/main/java/org/apache/qpid/management/jmx/MessageStatisticsTest.java
@@ -66,7 +66,7 @@ public class MessageStatisticsTest extends MessageStatisticsTestCase
             assertEquals("Incorrect connection total", 45, total);
             assertEquals("Incorrect connection data", 4500, data);
         }
-        if (!_broker.equals(VM))
+        if (!_brokerType.equals(VM))
         {
             assertEquals("Incorrect server total", 45, _jmxUtils.getServerInformation().getTotalMessagesReceived());
             assertEquals("Incorrect server data", 4500, _jmxUtils.getServerInformation().getTotalDataReceived());
@@ -147,7 +147,7 @@ public class MessageStatisticsTest extends MessageStatisticsTestCase
             assertEquals("Incorrect active connection total", 20, total);
             assertEquals("Incorrect active connection data", 2000, data);
         }
-        if (!_broker.equals(VM))
+        if (!_brokerType.equals(VM))
         {
             assertEquals("Incorrect server total", 30, _jmxUtils.getServerInformation().getTotalMessagesReceived());
             assertEquals("Incorrect server data", 3000, _jmxUtils.getServerInformation().getTotalDataReceived());
@@ -187,7 +187,7 @@ public class MessageStatisticsTest extends MessageStatisticsTestCase
         assertApprox("Incorrect dev vhost peak messages", 0.2d, 2.0d, dev.getPeakMessageReceiptRate());
         assertApprox("Incorrect dev vhost peak data", 0.2d, 20.0d, dev.getPeakDataReceiptRate());
 
-        if (!_broker.equals(VM))
+        if (!_brokerType.equals(VM))
         {
             assertApprox("Incorrect server peak messages", 0.2d, 2.0d, _jmxUtils.getServerInformation().getPeakMessageReceiptRate());
             assertApprox("Incorrect server peak data", 0.2d, 20.0d, _jmxUtils.getServerInformation().getPeakDataReceiptRate());
@@ -211,7 +211,7 @@ public class MessageStatisticsTest extends MessageStatisticsTestCase
         assertEquals("Incorrect dev vhost total messages", 10, dev.getTotalMessagesReceived());
         assertEquals("Incorrect dev vhost total data", 100, dev.getTotalDataReceived());
 
-        if (!_broker.equals(VM))
+        if (!_brokerType.equals(VM))
         {
             assertEquals("Incorrect server total messages", 20, _jmxUtils.getServerInformation().getTotalMessagesReceived());
             assertEquals("Incorrect server total data", 200, _jmxUtils.getServerInformation().getTotalDataReceived());
@@ -224,7 +224,7 @@ public class MessageStatisticsTest extends MessageStatisticsTestCase
         assertEquals("Incorrect dev vhost total messages", 10, dev.getTotalMessagesReceived());
         assertEquals("Incorrect dev vhost total data", 100, dev.getTotalDataReceived());
 
-        if (!_broker.equals(VM))
+        if (!_brokerType.equals(VM))
         {
             assertEquals("Incorrect server total messages", 20, _jmxUtils.getServerInformation().getTotalMessagesReceived());
             assertEquals("Incorrect server total data", 200, _jmxUtils.getServerInformation().getTotalDataReceived());
diff --git a/java/systests/src/main/java/org/apache/qpid/server/BrokerStartupTest.java b/java/systests/src/main/java/org/apache/qpid/server/BrokerStartupTest.java
index f9227c53ba..27b4de0a8e 100644
--- a/java/systests/src/main/java/org/apache/qpid/server/BrokerStartupTest.java
+++ b/java/systests/src/main/java/org/apache/qpid/server/BrokerStartupTest.java
@@ -70,13 +70,13 @@ public class BrokerStartupTest extends AbstractTestLogging
     {
         // This logging startup code only occurs when you run a Java broker,
         // that broker must be started via Main so not an InVM broker.
-        if (isJavaBroker() && isExternalBroker())
+        if (isJavaBroker() && isExternalBroker() && !isInternalBroker())
         {
             //Remove test Log4j config from the commandline
-            _broker = _broker.substring(0, _broker.indexOf("-l"));
+            _brokerCommand = _brokerCommand.substring(0, _brokerCommand.indexOf("-l"));
 
             // Add an invalid value
-            _broker += " -l invalid";
+            _brokerCommand += " -l invalid";
 
             // The  broker has a built in default log4j configuration set up
             // so if the the broker cannot load the -l value it will use default
diff --git a/java/systests/src/main/java/org/apache/qpid/server/logging/BrokerLoggingTest.java b/java/systests/src/main/java/org/apache/qpid/server/logging/BrokerLoggingTest.java
index 8fd2c085c3..9155b84365 100644
--- a/java/systests/src/main/java/org/apache/qpid/server/logging/BrokerLoggingTest.java
+++ b/java/systests/src/main/java/org/apache/qpid/server/logging/BrokerLoggingTest.java
@@ -21,6 +21,8 @@
 package org.apache.qpid.server.logging;
 
 import junit.framework.AssertionFailedError;
+
+import org.apache.qpid.server.BrokerOptions;
 import org.apache.qpid.server.Main;
 import org.apache.qpid.transport.ConnectionException;
 import org.apache.qpid.util.LogMonitor;
@@ -151,12 +153,12 @@ public class BrokerLoggingTest extends AbstractTestLogging
     {
         // This logging startup code only occurs when you run a Java broker,
         // that broker must be started via Main so not an InVM broker.
-        if (isJavaBroker() && isExternalBroker())
+        if (isJavaBroker() && isExternalBroker() && !isInternalBroker())
         {
             String TESTID = "BRK-1007";
 
             //Remove test Log4j config from the commandline
-            _broker = _broker.substring(0, _broker.indexOf("-l"));
+            _brokerCommand = _brokerCommand.substring(0, _brokerCommand.indexOf("-l"));
 
             startBroker();
 
@@ -203,7 +205,7 @@ public class BrokerLoggingTest extends AbstractTestLogging
                                  1, findMatches(TESTID).size());
 
                     //3
-                    String defaultLog4j = _configFile.getParent() + "/" + Main.DEFAULT_LOG_CONFIG_FILENAME;
+                    String defaultLog4j = System.getProperty(QPID_HOME) + "/" + BrokerOptions.DEFAULT_LOG_CONFIG_FILE;
                     assertTrue("Log4j file(" + defaultLog4j + ") details not correctly logged:" + getMessageString(log),
                                getMessageString(log).endsWith(defaultLog4j));
 
@@ -240,12 +242,11 @@ public class BrokerLoggingTest extends AbstractTestLogging
      */
     public void testBrokerStartupCustomLog4j() throws Exception
     {
-        // This logging startup code only occurs when you run a Java broker,
-        // that broker must be started via Main so not an InVM broker.
+        // This logging startup code only occurs when you run a Java broker
         if (isJavaBroker() && isExternalBroker())
         {
             // Get custom -l value used during testing for the broker startup
-            String customLog4j = _broker.substring(_broker.indexOf("-l") + 2);
+            String customLog4j = _brokerCommand.substring(_brokerCommand.indexOf("-l") + 2);
 
             String TESTID = "BRK-1007";
 
diff --git a/java/systests/src/main/java/org/apache/qpid/server/persistent/NoLocalAfterRecoveryTest.java b/java/systests/src/main/java/org/apache/qpid/server/persistent/NoLocalAfterRecoveryTest.java
index a5aec3edce..ce6cc60ffd 100644
--- a/java/systests/src/main/java/org/apache/qpid/server/persistent/NoLocalAfterRecoveryTest.java
+++ b/java/systests/src/main/java/org/apache/qpid/server/persistent/NoLocalAfterRecoveryTest.java
@@ -80,7 +80,7 @@ public class NoLocalAfterRecoveryTest extends QpidBrokerTestCase implements Conn
 
         // Due to the problem with SingleServer delaying on all connection
         // attempts. So using a high retry value.
-        if (_broker.equals(VM))
+        if (_brokerType.equals(VM))
         {
             // Local testing suggests InVM restart takes under a second
             details.setProperty(BrokerDetails.OPTIONS_RETRY, "5");
diff --git a/java/systests/src/main/java/org/apache/qpid/server/security/firewall/FirewallConfigTest.java b/java/systests/src/main/java/org/apache/qpid/server/security/firewall/FirewallConfigTest.java
index f40e95885d..be4a865ff6 100644
--- a/java/systests/src/main/java/org/apache/qpid/server/security/firewall/FirewallConfigTest.java
+++ b/java/systests/src/main/java/org/apache/qpid/server/security/firewall/FirewallConfigTest.java
@@ -85,7 +85,7 @@ public class FirewallConfigTest extends QpidBrokerTestCase
 
     public void testVhostAllowBrokerDeny() throws Exception
     {
-        if (_broker.equals(VM))
+        if (_brokerType.equals(VM))
         {
             //No point running this test with an InVM broker as the
             //firewall plugin only functions for TCP connections.
@@ -125,7 +125,7 @@ public class FirewallConfigTest extends QpidBrokerTestCase
     
     public void testVhostDenyBrokerAllow() throws Exception
     {
-        if (_broker.equals(VM))
+        if (_brokerType.equals(VM))
         {
             //No point running this test with an InVM broker as the
             //firewall plugin only functions for TCP connections.
@@ -277,7 +277,7 @@ public class FirewallConfigTest extends QpidBrokerTestCase
 
     private void testFirewall(boolean initial, boolean inVhost, Runnable restartOrReload) throws Exception
     {
-        if (_broker.equals(VM))
+        if (_brokerType.equals(VM))
         {
             // No point running this test in a vm broker
             return;
diff --git a/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutTestCase.java b/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutTestCase.java
index 637f43fb2c..5fb06dfbb4 100644
--- a/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutTestCase.java
+++ b/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutTestCase.java
@@ -102,7 +102,7 @@ public class TransactionTimeoutTestCase extends QpidBrokerTestCase implements Ex
         super.setUp();
         
         // Connect to broker
-        String broker = _broker.equals(VM) ? ("vm://:" + DEFAULT_VM_PORT) : ("tcp://localhost:" + DEFAULT_PORT);
+        String broker = _brokerType.equals(VM) ? ("vm://:" + DEFAULT_VM_PORT) : ("tcp://localhost:" + DEFAULT_PORT);
         ConnectionURL url = new AMQConnectionURL("amqp://guest:guest@clientid/test?brokerlist='" + broker + "'&maxprefetch='1'");
         _con = (AMQConnection) getConnection(url);
         _con.setExceptionListener(this);
diff --git a/java/systests/src/main/java/org/apache/qpid/test/utils/BrokerHolder.java b/java/systests/src/main/java/org/apache/qpid/test/utils/BrokerHolder.java
new file mode 100644
index 0000000000..8345803d56
--- /dev/null
+++ b/java/systests/src/main/java/org/apache/qpid/test/utils/BrokerHolder.java
@@ -0,0 +1,26 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.test.utils;
+
+public interface BrokerHolder
+{
+    void shutdown();
+}
diff --git a/java/systests/src/main/java/org/apache/qpid/test/utils/FailoverBaseCase.java b/java/systests/src/main/java/org/apache/qpid/test/utils/FailoverBaseCase.java
index d3b429e315..1832a7641e 100644
--- a/java/systests/src/main/java/org/apache/qpid/test/utils/FailoverBaseCase.java
+++ b/java/systests/src/main/java/org/apache/qpid/test/utils/FailoverBaseCase.java
@@ -34,24 +34,10 @@ public class FailoverBaseCase extends QpidBrokerTestCase
 {
     protected static final Logger _logger = LoggerFactory.getLogger(FailoverBaseCase.class);
 
-    public static int FAILING_VM_PORT = 2;
-    public static int FAILING_PORT = Integer.parseInt(System.getProperty("test.port.alt"));
     public static final long DEFAULT_FAILOVER_TIME = 10000L;
 
     protected int failingPort;
 
-    protected int getFailingPort()
-    {
-        if (_broker.equals(VM))
-        {
-            return FAILING_VM_PORT;
-        }
-        else
-        {
-        	return FAILING_PORT;
-        }
-    }
-
     protected void setUp() throws java.lang.Exception
     {
         super.setUp();
diff --git a/java/systests/src/main/java/org/apache/qpid/test/utils/InternalBrokerHolder.java b/java/systests/src/main/java/org/apache/qpid/test/utils/InternalBrokerHolder.java
new file mode 100644
index 0000000000..340f00fed8
--- /dev/null
+++ b/java/systests/src/main/java/org/apache/qpid/test/utils/InternalBrokerHolder.java
@@ -0,0 +1,50 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.test.utils;
+
+import org.apache.log4j.Logger;
+import org.apache.qpid.server.Broker;
+
+public class InternalBrokerHolder implements BrokerHolder
+{
+    private static final Logger LOGGER = Logger.getLogger(InternalBrokerHolder.class);
+    private final Broker _broker;
+
+    public InternalBrokerHolder(final Broker broker)
+    {
+        if(broker == null)
+        {
+            throw new IllegalArgumentException("Broker must not be null");
+        }
+
+        _broker = broker;
+    }
+
+    public void shutdown()
+    {
+        LOGGER.info("Shutting down Broker instance");
+
+        _broker.shutdown();
+        
+        LOGGER.info("Broker instance shutdown");
+    }
+
+}
diff --git a/java/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java b/java/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java
index a5d1f4821f..66c8a60c36 100644
--- a/java/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java
+++ b/java/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java
@@ -25,6 +25,7 @@ import java.io.InputStreamReader;
 import java.io.LineNumberReader;
 import java.io.PrintStream;
 import java.net.MalformedURLException;
+import java.net.URL;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -62,6 +63,9 @@ import org.apache.qpid.exchange.ExchangeDefaults;
 import org.apache.qpid.jms.BrokerDetails;
 import org.apache.qpid.jms.ConnectionURL;
 import org.apache.qpid.management.common.mbeans.ConfigurationManagement;
+import org.apache.qpid.server.Broker;
+import org.apache.qpid.server.BrokerOptions;
+import org.apache.qpid.server.ProtocolExclusion;
 import org.apache.qpid.server.configuration.ServerConfiguration;
 import org.apache.qpid.server.registry.ApplicationRegistry;
 import org.apache.qpid.server.registry.ConfigurationFileApplicationRegistry;
@@ -107,7 +111,8 @@ public class QpidBrokerTestCase extends QpidTestCase
 
     // system properties
     private static final String BROKER_LANGUAGE = "broker.language";
-    private static final String BROKER = "broker";
+    private static final String BROKER_TYPE = "broker.type";
+    private static final String BROKER_COMMAND = "broker.command";
     private static final String BROKER_CLEAN = "broker.clean";
     private static final String BROKER_CLEAN_BETWEEN_TESTS = "broker.clean.between.tests";
     private static final String BROKER_EXISTING_QPID_WORK = "broker.existing.qpid.work";
@@ -118,12 +123,15 @@ public class QpidBrokerTestCase extends QpidTestCase
     private static final String BROKER_LOG_INTERLEAVE = "broker.log.interleave";
     private static final String BROKER_LOG_PREFIX = "broker.log.prefix";
     private static final String BROKER_PERSITENT = "broker.persistent";
+    private static final String BROKER_PROTOCOL_EXCLUDES = "broker.protocols.excludes";
+    
 
     // values
     protected static final String JAVA = "java";
     protected static final String CPP = "cpp";
     protected static final String VM = "vm";
     protected static final String EXTERNAL = "external";
+    protected static final String INTERNAL = "internal";
     private static final String VERSION_08 = "0-8";
     private static final String VERSION_010 = "0-10";
 
@@ -131,16 +139,19 @@ public class QpidBrokerTestCase extends QpidTestCase
 
     public static final int DEFAULT_VM_PORT = 1;
     public static final int DEFAULT_PORT = Integer.getInteger("test.port", ServerConfiguration.DEFAULT_PORT);
+    public static final int FAILING_PORT = Integer.parseInt(System.getProperty("test.port.alt"));
     public static final int DEFAULT_MANAGEMENT_PORT = Integer.getInteger("test.mport", ServerConfiguration.DEFAULT_JMXPORT);
     public static final int DEFAULT_SSL_PORT = Integer.getInteger("test.sslport", ServerConfiguration.DEFAULT_SSL_PORT);
 
     protected String _brokerLanguage = System.getProperty(BROKER_LANGUAGE, JAVA);
-    protected String _broker = System.getProperty(BROKER, VM);
+    protected String _brokerType = System.getProperty(BROKER_TYPE, INTERNAL);
+    protected String _brokerCommand = System.getProperty(BROKER_COMMAND);
     private String _brokerClean = System.getProperty(BROKER_CLEAN, null);
     private Boolean _brokerCleanBetweenTests = Boolean.getBoolean(BROKER_CLEAN_BETWEEN_TESTS);
     private String _brokerVersion = System.getProperty(BROKER_VERSION, VERSION_08);
     protected String _output = System.getProperty(TEST_OUTPUT);
     protected Boolean _brokerPersistent = Boolean.getBoolean(BROKER_PERSITENT);
+    private String _brokerProtocolExcludes = System.getProperty(BROKER_PROTOCOL_EXCLUDES);
 
     protected static String _brokerLogPrefix = System.getProperty(BROKER_LOG_PREFIX,"BROKER: ");
     protected static boolean _interleaveBrokerLog = Boolean.getBoolean(BROKER_LOG_INTERLEAVE);
@@ -149,7 +160,7 @@ public class QpidBrokerTestCase extends QpidTestCase
 
     protected PrintStream _brokerOutputStream;
 
-    protected Map<Integer, Process> _brokers = new HashMap<Integer, Process>();
+    protected Map<Integer, BrokerHolder> _brokers = new HashMap<Integer, BrokerHolder>();
 
     protected InitialContext _initialContext;
     protected AMQConnectionFactory _connectionFactory;
@@ -291,7 +302,7 @@ public class QpidBrokerTestCase extends QpidTestCase
             cleanBroker();
 
             File existing = new File(existingQpidWorkPath);
-            File qpidWork = new File(getQpidWork(_broker, getPort()));
+            File qpidWork = new File(getQpidWork(_brokerType, getPort()));
             FileUtils.copyRecursive(existing, qpidWork);
         }
 
@@ -395,11 +406,6 @@ public class QpidBrokerTestCase extends QpidTestCase
         }
     }
 
-    public void startBroker() throws Exception
-    {
-        startBroker(0);
-    }
-
     /**
      * Return the management portin use by the broker on this main port
      *
@@ -409,7 +415,7 @@ public class QpidBrokerTestCase extends QpidTestCase
      */
     protected int getManagementPort(int mainPort)
     {
-        return mainPort + (DEFAULT_MANAGEMENT_PORT - (_broker.equals(VM) ? DEFAULT_VM_PORT : DEFAULT_PORT));
+        return mainPort + (DEFAULT_MANAGEMENT_PORT - (_brokerType.equals(VM) ? DEFAULT_VM_PORT : DEFAULT_PORT));
     }
 
     /**
@@ -424,11 +430,11 @@ public class QpidBrokerTestCase extends QpidTestCase
 
     protected int getPort(int port)
     {
-        if (_broker.equals(VM))
+        if (_brokerType.equals(VM))
         {
             return port == 0 ? DEFAULT_VM_PORT : port;
         }
-        else if (!_broker.equals(EXTERNAL))
+        else if (!_brokerType.equals(EXTERNAL))
         {
             return port == 0 ? DEFAULT_PORT : port;
         }
@@ -440,11 +446,18 @@ public class QpidBrokerTestCase extends QpidTestCase
 
     protected String getBrokerCommand(int port) throws MalformedURLException
     {
-        return _broker
+        final String protocolExcludesList = _brokerProtocolExcludes.replace("@PORT", "" + port);
+        return _brokerCommand
                 .replace("@PORT", "" + port)
                 .replace("@SSL_PORT", "" + (port - 1))
                 .replace("@MPORT", "" + getManagementPort(port))
-                .replace("@CONFIG_FILE", _configFile.toString());
+                .replace("@CONFIG_FILE", _configFile.toString())
+                .replace("@EXCLUDES", protocolExcludesList);
+    }
+
+    public void startBroker() throws Exception
+    {
+        startBroker(0);
     }
 
     public void startBroker(int port) throws Exception
@@ -455,8 +468,12 @@ public class QpidBrokerTestCase extends QpidTestCase
         saveTestConfiguration();
         saveTestVirtualhosts();
 
-        Process process = null;
-        if (_broker.equals(VM))
+        if(_brokers.get(port) != null)
+        {
+            throw new IllegalStateException("There is already an existing broker running on port " + port);
+        }
+
+        if (_brokerType.equals(VM))
         {
             setConfigurationProperty("management.jmxport", String.valueOf(getManagementPort(port)));
             setConfigurationProperty(ServerConfiguration.MGMT_CUSTOM_REGISTRY_SOCKET, String.valueOf(false));
@@ -483,10 +500,33 @@ public class QpidBrokerTestCase extends QpidTestCase
             }
             TransportConnection.createVMBroker(port);
         }
-        else if (!_broker.equals(EXTERNAL))
+        else if (_brokerType.equals(INTERNAL) && !existingInternalBroker())
+        {
+            setConfigurationProperty(ServerConfiguration.MGMT_CUSTOM_REGISTRY_SOCKET, String.valueOf(false));
+            saveTestConfiguration();
+
+            BrokerOptions options = new BrokerOptions();
+            options.setConfigFile(_configFile.getAbsolutePath());
+            options.addPort(port);
+
+            addExcludedPorts(port, options);
+
+            options.setJmxPort(getManagementPort(port));
+
+            //Set the log config file, relying on the log4j.configuration system property
+            //set on the JVM by the JUnit runner task in module.xml.
+            options.setLogConfigFile(new URL(System.getProperty("log4j.configuration")).getFile());
+
+            Broker broker = new Broker();
+            _logger.info("starting internal broker (same JVM)");
+            broker.startup(options);
+
+            _brokers.put(port, new InternalBrokerHolder(broker));
+        }
+        else if (!_brokerType.equals(EXTERNAL))
         {
             String cmd = getBrokerCommand(port);
-            _logger.info("starting broker: " + cmd);
+            _logger.info("starting external broker: " + cmd);
             ProcessBuilder pb = new ProcessBuilder(cmd.split("\\s+"));
             pb.redirectErrorStream(true);
 
@@ -502,7 +542,7 @@ public class QpidBrokerTestCase extends QpidTestCase
             // DON'T change PNAME, qpid.stop needs this value.
             env.put("QPID_PNAME", "-DPNAME=QPBRKR -DTNAME=\"" + _testName + "\"");
             // Add the port to QPID_WORK to ensure unique working dirs for multi broker tests
-            env.put("QPID_WORK", getQpidWork(_broker, port));
+            env.put("QPID_WORK", getQpidWork(_brokerType, port));
 
 
             // Use the environment variable to set amqj.logging.level for the broker
@@ -554,8 +594,7 @@ public class QpidBrokerTestCase extends QpidTestCase
                     env.put("QPID_OPTS", QPID_OPTS);
                 }
             }
-
-            process = pb.start();
+            Process process = pb.start();;
 
             Piper p = new Piper(process.getInputStream(),
             		            _brokerOutputStream,
@@ -576,6 +615,7 @@ public class QpidBrokerTestCase extends QpidTestCase
 
             try
             {
+                //test that the broker is still running and hasn't exited unexpectedly
                 int exit = process.exitValue();
                 _logger.info("broker aborted: " + exit);
                 cleanBroker();
@@ -583,11 +623,43 @@ public class QpidBrokerTestCase extends QpidTestCase
             }
             catch (IllegalThreadStateException e)
             {
-                // this is expect if the broker started succesfully
+                // this is expect if the broker started successfully
+            }
+
+            _brokers.put(port, new SpawnedBrokerHolder(process));
+        }
+    }
+
+    private void addExcludedPorts(int port, BrokerOptions options)
+    {
+        final String protocolExcludesList = _brokerProtocolExcludes.replace("@PORT", "" + port);
+        final String[] toks = protocolExcludesList.split("\\s");
+
+        if(toks.length % 2 != 0)
+        {
+            throw new IllegalArgumentException("Must be an even number of tokens in " + protocolExcludesList);
+        }
+        for (int i = 0; i < toks.length; i=i+2)
+        {
+            String excludeArg = toks[i];
+            final int excludedPort = Integer.parseInt(toks[i+1]);
+            options.addExcludedPort(ProtocolExclusion.lookup(excludeArg), excludedPort);
+
+            _logger.info("Adding protocol exclusion " + excludeArg + " " + excludedPort);
+        }
+    }
+
+    private boolean existingInternalBroker()
+    {
+        for(BrokerHolder holder : _brokers.values())
+        {
+            if(holder instanceof InternalBrokerHolder)
+            {
+                return true;
             }
         }
 
-        _brokers.put(port, process);
+        return false;
     }
 
     private String getQpidWork(String broker, int port)
@@ -682,14 +754,12 @@ public class QpidBrokerTestCase extends QpidTestCase
         port = getPort(port);
 
         _logger.info("stopping broker: " + getBrokerCommand(port));
-        Process process = _brokers.remove(port);
-        if (process != null)
+        BrokerHolder broker = _brokers.remove(port);
+        if (broker != null)
         {
-            process.destroy();
-            process.waitFor();
-            _logger.info("broker exited: " + process.exitValue());
+            broker.shutdown();
         }
-        else if (_broker.equals(VM))
+        else if (_brokerType.equals(VM))
         {
             TransportConnection.killVMBroker(port);
             ApplicationRegistry.remove();
@@ -940,7 +1010,7 @@ public class QpidBrokerTestCase extends QpidTestCase
 
     protected boolean isJavaBroker()
     {
-        return _brokerLanguage.equals("java") || _broker.equals("vm");
+        return _brokerLanguage.equals("java") || _brokerType.equals("vm");
     }
 
     protected boolean isCppBroker()
@@ -950,9 +1020,14 @@ public class QpidBrokerTestCase extends QpidTestCase
 
     protected boolean isExternalBroker()
     {
-        return !_broker.equals("vm");
+        return !_brokerType.equals("vm");
     }
-    
+
+    protected boolean isInternalBroker()
+    {
+        return _brokerType.equals(INTERNAL);
+    }
+
     protected boolean isBrokerStorePersistent()
     {
         return _brokerPersistent;
@@ -1024,7 +1099,7 @@ public class QpidBrokerTestCase extends QpidTestCase
      */
     public AMQConnectionFactory getConnectionFactory(String factoryName) throws NamingException
     {
-        if (_broker.equals(VM))
+        if (_brokerType.equals(VM))
         {
             factoryName += ".vm";
         }
@@ -1070,7 +1145,7 @@ public class QpidBrokerTestCase extends QpidTestCase
     {
         _logger.info("get Connection");
         Connection con;
-        if (_broker.equals(VM))
+        if (_brokerType.equals(VM))
         {
             con = new AMQConnection("vm://:1", username, password, id, "test");
         }
@@ -1335,7 +1410,7 @@ public class QpidBrokerTestCase extends QpidTestCase
      */
     public void reloadBrokerSecurityConfig() throws Exception
     {
-        if (_broker.equals(VM))
+        if (_brokerType.equals(VM))
         {
             ApplicationRegistry.getInstance().getConfiguration().reparseConfigFileSecuritySections();
         }
@@ -1360,4 +1435,16 @@ public class QpidBrokerTestCase extends QpidTestCase
 
         }
     }
+
+    protected int getFailingPort()
+    {
+        if (_brokerType.equals(VM))
+        {
+            throw new RuntimeException("VM is not supported for Failover testing");
+        }
+        else
+        {
+        	return FAILING_PORT;
+        }
+    }
 }
diff --git a/java/systests/src/main/java/org/apache/qpid/test/utils/SpawnedBrokerHolder.java b/java/systests/src/main/java/org/apache/qpid/test/utils/SpawnedBrokerHolder.java
new file mode 100644
index 0000000000..65239bbe02
--- /dev/null
+++ b/java/systests/src/main/java/org/apache/qpid/test/utils/SpawnedBrokerHolder.java
@@ -0,0 +1,58 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.test.utils;
+
+import org.apache.log4j.Logger;
+
+public class SpawnedBrokerHolder implements BrokerHolder
+{
+    private static final Logger LOGGER = Logger.getLogger(SpawnedBrokerHolder.class);
+
+    private final Process _process;
+
+    public SpawnedBrokerHolder(final Process process)
+    {
+        if(process == null)
+        {
+            throw new IllegalArgumentException("Process must not be null");
+        }
+
+        _process = process;
+    }
+
+    public void shutdown()
+    {
+        LOGGER.info("Destroying broker process");
+
+        _process.destroy();
+
+        try
+        {
+            _process.waitFor();
+            LOGGER.info("broker exited: " + _process.exitValue());
+        }
+        catch (InterruptedException e)
+        {
+            LOGGER.error("Interrupted whilst waiting for process destruction");
+            Thread.currentThread().interrupt();
+        }
+    }
+}
diff --git a/java/test-profiles/default.testprofile b/java/test-profiles/default.testprofile
index df8148f787..85ab263f82 100644
--- a/java/test-profiles/default.testprofile
+++ b/java/test-profiles/default.testprofile
@@ -20,11 +20,9 @@ java.naming.factory.initial=org.apache.qpid.jndi.PropertiesFileInitialContextFac
 java.naming.provider.url=${test.profiles}/test-provider.properties
 
 broker.version=0-8
-broker=vm
+broker.type=vm
 broker.clean=${test.profiles}/clean-dir ${build.data} ${project.root}/build/work
 broker.ready=Listening on TCP port
-broker.start=${test.profiles}/start-broker
-broker.kill=${test.profiles}/kill-broker
 broker.config=${project.root}/build/etc/config-systests.xml
 messagestore.class.name=org.apache.qpid.server.store.MemoryMessageStore
 
diff --git a/java/test-profiles/internal.testprofile b/java/test-profiles/internal.testprofile
new file mode 100644
index 0000000000..df544b1a20
--- /dev/null
+++ b/java/test-profiles/internal.testprofile
@@ -0,0 +1,32 @@
+#
+# 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.
+#
+broker.language=java
+broker.type=internal
+#broker.command only used for the second broker during failover tests in this profile
+broker.command=${project.root}/build/bin/qpid-server -p @PORT -m @MPORT @EXCLUDES -c @CONFIG_FILE -l ${test.profiles}/log4j-test.xml
+#broker.command=${project.root}/build/bin/qpid-server -p @PORT -m @MPORT --exclude-0-10 @PORT -c @CONFIG_FILE -l ${test.profiles}/log4j-test.xml
+broker.clean=${test.profiles}/clean-dir ${build.data} ${project.root}/build/work
+broker.ready=BRK-1004
+broker.stopped=Exception
+broker.protocols.excludes=--exclude-0-10 @PORT
+#
+# Do not enable. Allow client to attempt 0-10 and negotiate downwards 
+#
+#qpid.amqp.version=0-91
+profile.excludes=JavaTransientExcludes JavaStandaloneExcludes 08StandaloneExcludes
diff --git a/java/test-profiles/java.testprofile b/java/test-profiles/java.testprofile
index c8c776d3e1..00d486309f 100644
--- a/java/test-profiles/java.testprofile
+++ b/java/test-profiles/java.testprofile
@@ -17,7 +17,8 @@
 # under the License.
 #
 broker.language=java
-broker=${project.root}/build/bin/qpid-server -p @PORT -m @MPORT --exclude-0-10 @PORT -c @CONFIG_FILE -l ${test.profiles}/log4j-test.xml
+broker.type=spawned
+broker.command=${project.root}/build/bin/qpid-server -p @PORT -m @MPORT --exclude-0-10 @PORT -c @CONFIG_FILE -l ${test.profiles}/log4j-test.xml
 broker.clean=${test.profiles}/clean-dir ${build.data} ${project.root}/build/work
 broker.ready=BRK-1004
 broker.stopped=Exception