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

4 files changed, 96 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
      */