QPID-3800: add the ability to disable particular protocol versions broker-wide to save having to exclude them from individual ports

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

7 files changed, 220 insertions, 13 deletions

diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/Broker.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/Broker.java
index cf5b57e803..a5dc3bb535 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/Broker.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/Broker.java
@@ -192,7 +192,7 @@ public class Broker
                 for(int port : ports)
                 {
                     final Set<AmqpProtocolVersion> supported =
-                                    getSupportedVersions(port, exclude_0_10, exclude_0_9_1, exclude_0_9, exclude_0_8);
+                                    getSupportedVersions(port, exclude_0_10, exclude_0_9_1, exclude_0_9, exclude_0_8, serverConfig);
                     final NetworkTransportConfiguration settings =
                                     new ServerNetworkTransportConfiguration(serverConfig, port, bindAddress.getHostName(), Transport.TCP);
 
@@ -217,7 +217,7 @@ public class Broker
                 for(int sslPort : sslPorts)
                 {
                     final Set<AmqpProtocolVersion> supported =
-                                    getSupportedVersions(sslPort, exclude_0_10, exclude_0_9_1, exclude_0_9, exclude_0_8);
+                                    getSupportedVersions(sslPort, exclude_0_10, exclude_0_9_1, exclude_0_9, exclude_0_8, serverConfig);
                     final NetworkTransportConfiguration settings =
                         new ServerNetworkTransportConfiguration(serverConfig, sslPort, bindAddress.getHostName(), Transport.TCP);
 
@@ -243,23 +243,24 @@ public class Broker
 
     private static Set<AmqpProtocolVersion> getSupportedVersions(final int port, final Set<Integer> exclude_0_10,
                                                                 final Set<Integer> exclude_0_9_1, final Set<Integer> exclude_0_9,
-                                                                final Set<Integer> exclude_0_8)
+                                                                final Set<Integer> exclude_0_8,
+                                                                final ServerConfiguration serverConfig)
     {
         final EnumSet<AmqpProtocolVersion> supported = EnumSet.allOf(AmqpProtocolVersion.class);
 
-        if(exclude_0_10.contains(port))
+        if(exclude_0_10.contains(port) || !serverConfig.isAmqp010enabled())
         {
             supported.remove(AmqpProtocolVersion.v0_10);
         }
-        if(exclude_0_9_1.contains(port))
+        if(exclude_0_9_1.contains(port) || !serverConfig.isAmqp091enabled())
         {
             supported.remove(AmqpProtocolVersion.v0_9_1);
         }
-        if(exclude_0_9.contains(port))
+        if(exclude_0_9.contains(port) || !serverConfig.isAmqp09enabled())
         {
             supported.remove(AmqpProtocolVersion.v0_9);
         }
-        if(exclude_0_8.contains(port))
+        if(exclude_0_8.contains(port) || !serverConfig.isAmqp08enabled())
         {
             supported.remove(AmqpProtocolVersion.v0_8);
         }
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java
index d81116e454..c2425b6794 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java
@@ -87,6 +87,10 @@ public class ServerConfiguration extends ConfigurationPlugin
     public static final String MGMT_JMXPORT_CONNECTORSERVER = "management.jmxport.connectorServer";
     public static final String STATUS_UPDATES = "status-updates";
     public static final String ADVANCED_LOCALE = "advanced.locale";
+    public static final String CONNECTOR_AMQP010ENABLED = "connector.amqp010enabled";
+    public static final String CONNECTOR_AMQP091ENABLED = "connector.amqp091enabled";
+    public static final String CONNECTOR_AMQP09ENABLED = "connector.amqp09enabled";
+    public static final String CONNECTOR_AMQP08ENABLED = "connector.amqp08enabled";
 
     {
         envVarMap.put("QPID_PORT", "connector.port");
@@ -837,4 +841,24 @@ public class ServerConfiguration extends ConfigurationPlugin
         return getConfig().getString("deadLetterQueueSuffix", AMQQueueFactory.DEFAULT_DLQ_NAME_SUFFIX);
     }
 
+    public boolean isAmqp010enabled()
+    {
+        return getConfig().getBoolean(CONNECTOR_AMQP010ENABLED, true);
+    }
+
+    public boolean isAmqp091enabled()
+    {
+        return getConfig().getBoolean(CONNECTOR_AMQP091ENABLED, true);
+    }
+
+    public boolean isAmqp09enabled()
+    {
+        return getConfig().getBoolean(CONNECTOR_AMQP09ENABLED, true);
+    }
+
+    public boolean isAmqp08enabled()
+    {
+        return getConfig().getBoolean(CONNECTOR_AMQP08ENABLED, true);
+    }
+
 }
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngine.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngine.java
index bba8e83e14..e361bd3ece 100755
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngine.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngine.java
@@ -21,8 +21,13 @@
 package org.apache.qpid.server.protocol;
 
 
-import org.apache.log4j.Logger;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.util.Set;
 
+import org.apache.log4j.Logger;
+import org.apache.qpid.framing.ProtocolVersion;
 import org.apache.qpid.protocol.ServerProtocolEngine;
 import org.apache.qpid.server.registry.IApplicationRegistry;
 import org.apache.qpid.server.transport.ServerConnection;
@@ -30,11 +35,6 @@ import org.apache.qpid.transport.ConnectionDelegate;
 import org.apache.qpid.transport.Sender;
 import org.apache.qpid.transport.network.NetworkConnection;
 
-import java.net.InetSocketAddress;
-import java.net.SocketAddress;
-import java.nio.ByteBuffer;
-import java.util.Set;
-
 public class MultiVersionProtocolEngine implements ServerProtocolEngine
 {
     private static final Logger _logger = Logger.getLogger(MultiVersionProtocolEngine.class);
@@ -391,6 +391,7 @@ public class MultiVersionProtocolEngine implements ServerProtocolEngine
 
                 ServerProtocolEngine newDelegate = null;
                 byte[] newestSupported = null;
+                AmqpProtocolVersion newestSupportedVersion = null;
 
                 for(int i = 0; newDelegate == null && i < _creators.length; i++)
                 {
@@ -398,6 +399,7 @@ public class MultiVersionProtocolEngine implements ServerProtocolEngine
                     if(_supported.contains(_creators[i].getVersion()))
                     {
                         newestSupported = _creators[i].getHeaderIdentifier();
+                        newestSupportedVersion = _creators[i].getVersion();
                         byte[] compareBytes = _creators[i].getHeaderIdentifier();
                         boolean equal = true;
                         for(int j = 0; equal && j<compareBytes.length; j++)
@@ -414,6 +416,10 @@ public class MultiVersionProtocolEngine implements ServerProtocolEngine
                 // If no delegate is found then send back the most recent support protocol version id
                 if(newDelegate == null)
                 {
+                    if(_logger.isDebugEnabled())
+                    {
+                        _logger.debug("Unsupported protocol version requested, replying with: " + newestSupportedVersion);
+                    }
                     _sender.send(ByteBuffer.wrap(newestSupported));
                     _sender.flush();
 
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/configuration/ServerConfigurationTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/configuration/ServerConfigurationTest.java
index abdc7f2246..4caefc2f18 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/configuration/ServerConfigurationTest.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/configuration/ServerConfigurationTest.java
@@ -1536,6 +1536,58 @@ public class ServerConfigurationTest extends QpidTestCase
         assertTrue("C3P0 queue DLQ should be enabled, using broker default", c3p0.isDeadLetterQueueEnabled());
     }
 
+    public void testIsAmqp010enabled() throws ConfigurationException
+    {
+        // Check default
+        _serverConfig.initialise();
+        assertEquals(true, _serverConfig.isAmqp010enabled());
+
+        // Check value we set
+        _config.setProperty(ServerConfiguration.CONNECTOR_AMQP010ENABLED, false);
+        _serverConfig = new ServerConfiguration(_config);
+        _serverConfig.initialise();
+        assertEquals(false, _serverConfig.isAmqp010enabled());
+    }
+
+    public void testIsAmqp091enabled() throws ConfigurationException
+    {
+        // Check default
+        _serverConfig.initialise();
+        assertEquals(true, _serverConfig.isAmqp091enabled());
+
+        // Check value we set
+        _config.setProperty(ServerConfiguration.CONNECTOR_AMQP091ENABLED, false);
+        _serverConfig = new ServerConfiguration(_config);
+        _serverConfig.initialise();
+        assertEquals(false, _serverConfig.isAmqp091enabled());
+    }
+
+    public void testIsAmqp09enabled() throws ConfigurationException
+    {
+        // Check default
+        _serverConfig.initialise();
+        assertEquals(true, _serverConfig.isAmqp09enabled());
+
+        // Check value we set
+        _config.setProperty(ServerConfiguration.CONNECTOR_AMQP09ENABLED, false);
+        _serverConfig = new ServerConfiguration(_config);
+        _serverConfig.initialise();
+        assertEquals(false, _serverConfig.isAmqp09enabled());
+    }
+
+    public void testIsAmqp08enabled() throws ConfigurationException
+    {
+        // Check default
+        _serverConfig.initialise();
+        assertEquals(true, _serverConfig.isAmqp08enabled());
+
+        // Check value we set
+        _config.setProperty(ServerConfiguration.CONNECTOR_AMQP08ENABLED, false);
+        _serverConfig = new ServerConfiguration(_config);
+        _serverConfig.initialise();
+        assertEquals(false, _serverConfig.isAmqp08enabled());
+    }
+
     /**
      * Convenience method to output required security preamble for broker config
      */
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/server/DisablingProtocolsTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/server/DisablingProtocolsTest.java
new file mode 100644
index 0000000000..5b3ac8d745
--- /dev/null
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/server/DisablingProtocolsTest.java
@@ -0,0 +1,122 @@
+/*
+ *  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.qpid.client.AMQConnection;
+import org.apache.qpid.configuration.ClientProperties;
+import org.apache.qpid.framing.ProtocolVersion;
+import org.apache.qpid.test.utils.QpidBrokerTestCase;
+
+/**
+ * Tests to validate it is possible to disable support for particular protocol
+ * versions entirely, rather than selectively excluding them on particular ports.
+ */
+public class DisablingProtocolsTest extends QpidBrokerTestCase
+{
+    public void setUp() throws Exception
+    {
+        // No-op, we call super.setUp() from test methods after appropriate config overrides
+    }
+
+    /**
+     * Test that 0-10, 0-9-1, 0-9, and 0-8 support is present when no
+     * attempt has yet been made to disable them, and forcing the client
+     * to negotiate from a particular protocol version returns a connection
+     * using the expected protocol version.
+     */
+    public void testDefaultProtocolSupport() throws Exception
+    {
+        //Start the broker without modifying its supported protocols
+        super.setUp();
+
+        //Verify requesting a 0-10 connection works
+        setTestClientSystemProperty(ClientProperties.AMQP_VERSION, "0-10");
+        AMQConnection connection = (AMQConnection) getConnection();
+        assertEquals("Unexpected protocol version in use", ProtocolVersion.v0_10, connection.getProtocolVersion());
+        connection.close();
+
+        //Verify requesting a 0-91 connection works
+        setTestClientSystemProperty(ClientProperties.AMQP_VERSION, "0-9-1");
+        connection = (AMQConnection) getConnection();
+        assertEquals("Unexpected protocol version in use", ProtocolVersion.v0_91, connection.getProtocolVersion());
+        connection.close();
+
+        //Verify requesting a 0-9 connection works
+        setTestClientSystemProperty(ClientProperties.AMQP_VERSION, "0-9");
+        connection = (AMQConnection) getConnection();
+        assertEquals("Unexpected protocol version in use", ProtocolVersion.v0_9, connection.getProtocolVersion());
+        connection.close();
+
+        //Verify requesting a 0-8 connection works
+        setTestClientSystemProperty(ClientProperties.AMQP_VERSION, "0-8");
+        connection = (AMQConnection) getConnection();
+        assertEquals("Unexpected protocol version in use", ProtocolVersion.v8_0, connection.getProtocolVersion());
+        connection.close();
+    }
+
+    public void testDisabling010() throws Exception
+    {
+        //disable 0-10 support
+        setConfigurationProperty("connector.amqp010enabled", "false");
+
+        super.setUp();
+
+        //Verify initially requesting a 0-10 connection now negotiates a 0-91
+        //connection as the broker should reply with its highest supported protocol
+        setTestClientSystemProperty(ClientProperties.AMQP_VERSION, "0-10");
+        AMQConnection connection = (AMQConnection) getConnection();
+        assertEquals("Unexpected protocol version in use", ProtocolVersion.v0_91, connection.getProtocolVersion());
+        connection.close();
+    }
+
+    public void testDisabling091and010() throws Exception
+    {
+        //disable 0-91 and 0-10 support
+        setConfigurationProperty("connector.amqp010enabled", "false");
+        setConfigurationProperty("connector.amqp091enabled", "false");
+
+        super.setUp();
+
+        //Verify initially requesting a 0-10 connection now negotiates a 0-9
+        //connection as the broker should reply with its highest supported protocol
+        setTestClientSystemProperty(ClientProperties.AMQP_VERSION, "0-10");
+        AMQConnection connection = (AMQConnection) getConnection();
+        assertEquals("Unexpected protocol version in use", ProtocolVersion.v0_9, connection.getProtocolVersion());
+        connection.close();
+    }
+
+    public void testDisabling09and091and010() throws Exception
+    {
+        //disable 0-9, 0-91 and 0-10 support
+        setConfigurationProperty("connector.amqp09enabled", "false");
+        setConfigurationProperty("connector.amqp091enabled", "false");
+        setConfigurationProperty("connector.amqp010enabled", "false");
+
+        super.setUp();
+
+        //Verify initially requesting a 0-10 connection now negotiates a 0-8
+        //connection as the broker should reply with its highest supported protocol
+        setTestClientSystemProperty(ClientProperties.AMQP_VERSION, "0-10");
+        AMQConnection connection = (AMQConnection) getConnection();
+        assertEquals("Unexpected protocol version in use", ProtocolVersion.v8_0, connection.getProtocolVersion());
+        connection.close();
+    }
+}
\ No newline at end of file
diff --git a/qpid/java/test-profiles/CPPExcludes b/qpid/java/test-profiles/CPPExcludes
index 582a201044..61b76bbdfc 100755
--- a/qpid/java/test-profiles/CPPExcludes
+++ b/qpid/java/test-profiles/CPPExcludes
@@ -153,6 +153,7 @@ org.apache.qpid.test.unit.transacted.TransactionTimeoutTest#*
 // Java broker only
 org.apache.qpid.server.logging.management.LoggingManagementMBeanTest#*
 org.apache.qpid.server.management.AMQUserManagementMBeanTest#*
+org.apache.qpid.server.DisablingProtocolsTest#*
 
 // QPID-3133: On 0-10, the exception listener is currently not invoked when reconnection fails to occurs.
 org.apache.qpid.server.failover.FailoverMethodTest#*
diff --git a/qpid/java/test-profiles/JavaPre010Excludes b/qpid/java/test-profiles/JavaPre010Excludes
index 4599356e1b..5d46010649 100644
--- a/qpid/java/test-profiles/JavaPre010Excludes
+++ b/qpid/java/test-profiles/JavaPre010Excludes
@@ -24,6 +24,7 @@
 // These tests requires a broker capable of 0-8/0-9/0-9-1 and 0-10 concurrently
 org.apache.qpid.test.client.message.JMSDestinationTest#testReceiveResend
 org.apache.qpid.server.message.MessageProtocolConversionTest#*
+org.apache.qpid.server.DisablingProtocolsTest#*
 
 
 // QPID-2478 test fails when run against broker using 0-8/9