Handle URI query parameters in chain of responsibility

With fallback hook, by default empty.

References #672

Author: acogoluegnes

src/main/java/com/rabbitmq/client/ConnectionFactory.java

@@ -22,6 +22,8 @@
 import com.rabbitmq.client.impl.recovery.RetryHandler;
 import com.rabbitmq.client.impl.recovery.TopologyRecoveryFilter;
 
+import java.util.Map.Entry;
+import java.util.function.BiConsumer;
 import javax.net.SocketFactory;
 import javax.net.ssl.SSLContext;
 import javax.net.ssl.SSLSocketFactory;
@@ -382,6 +384,36 @@ private static String uriDecode(String s) {
         }
     }
 
+  private static final Map<String, BiConsumer<String, ConnectionFactory>> URI_QUERY_PARAMETER_HANDLERS =
+      new HashMap<String, BiConsumer<String, ConnectionFactory>>() {
+        {
+            put("heartbeat", (value, cf) -> {
+                try {
+                    int heartbeatInt = Integer.parseInt(value);
+                    cf.setRequestedHeartbeat(heartbeatInt);
+                } catch (NumberFormatException e) {
+                    throw new IllegalArgumentException("Requested heartbeat must an integer");
+                }
+            });
+            put("connection_timeout", (value, cf) -> {
+              try {
+                int connectionTimeoutInt = Integer.parseInt(value);
+                cf.setConnectionTimeout(connectionTimeoutInt);
+              } catch (NumberFormatException e) {
+                throw new IllegalArgumentException("TCP connection timeout must an integer");
+              }
+            });
+            put("channel_max", (value, cf) -> {
+                try {
+                    int channelMaxInt = Integer.parseInt(value);
+                    cf.setRequestedChannelMax(channelMaxInt);
+                } catch (NumberFormatException e) {
+                    throw new IllegalArgumentException("Requested channel max must an integer");
+                }
+            });
+        }
+      };
+
     /**
      * Convenience method for setting some fields from query parameters
      * Will handle only a subset of the query parameters supported by the
@@ -391,7 +423,6 @@ private static String uriDecode(String s) {
      */
     private void setQuery(String rawQuery) {
         Map<String, String> parameters = new HashMap<>();
-        
         // parsing the query parameters
         try {
             for (String param : rawQuery.split("&")) {
@@ -404,43 +435,31 @@ private void setQuery(String rawQuery) {
                 parameters.put(key, value);
             }
         } catch (IOException e) {
-            throw new RuntimeException("Cannot parse the query parameters", e);
+            throw new IllegalArgumentException("Cannot parse the query parameters", e);
         }
 
-        // heartbeat
-        String heartbeat = parameters.get("heartbeat");
-        if (heartbeat != null) {
-            try {
-                int heartbeatInt = Integer.parseInt(heartbeat);
-                setRequestedHeartbeat(heartbeatInt);
-            } catch (NumberFormatException e) {
-                throw new IllegalArgumentException("Requested heartbeat must an integer");
-            }
-        }
-        
-        // connection_timeout
-        String connectionTimeout = parameters.get("connection_timeout");
-        if (connectionTimeout != null) {
-            try {
-                int connectionTimeoutInt = Integer.parseInt(connectionTimeout);
-                setConnectionTimeout(connectionTimeoutInt);
-            } catch (NumberFormatException e) {
-                throw new IllegalArgumentException("TCP connection timeout must an integer");
-            }
-        }
-        
-        // channel_max
-        String channelMax = parameters.get("channel_max");
-        if (channelMax != null) {
-            try {
-                int channelMaxInt = Integer.parseInt(channelMax);
-                setRequestedChannelMax(channelMaxInt);
-            } catch (NumberFormatException e) {
-                throw new IllegalArgumentException("Requested channel max must an integer");
+        for (Entry<String, String> entry : parameters.entrySet()) {
+            BiConsumer<String, ConnectionFactory> handler = URI_QUERY_PARAMETER_HANDLERS
+                .get(entry.getKey());
+            if (handler != null) {
+                handler.accept(entry.getValue(), this);
+            } else {
+                processUriQueryParameter(entry.getKey(), entry.getValue());
             }
         }
     }
 
+    /**
+     * Hook to process query parameters not handled natively.
+     * Handled natively: <code>heartbeat</code>, <code>connection_timeout</code>,
+     * <code>channel_max</code>.
+     * @param key
+     * @param value
+     */
+    protected void processUriQueryParameter(String key, String value) {
+
+    }
+
     /**
      * Retrieve the requested maximum channel number
      * @return the initially requested maximum channel number; zero for unlimited

src/test/java/com/rabbitmq/client/test/AmqpUriTest.java

@@ -14,18 +14,21 @@
 // [email protected].
 package com.rabbitmq.client.test;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.fail;
 
 import java.net.URISyntaxException;
 import java.security.KeyManagementException;
 import java.security.NoSuchAlgorithmException;
 
+import java.util.HashMap;
+import java.util.Map;
 import org.junit.Test;
 
 import com.rabbitmq.client.ConnectionFactory;
 
-public class AmqpUriTest extends BrokerTestCase
+public class AmqpUriTest
 {
     @Test public void uriParsing()
         throws URISyntaxException, NoSuchAlgorithmException, KeyManagementException
@@ -109,6 +112,19 @@ public class AmqpUriTest extends BrokerTestCase
         parseFail("amqp://user:pass@host:10000/vhost?heartbeat=342?connection_timeout=442");
     }
 
+    @Test
+    public void processUriQueryParameterShouldBeCalledForNotHandledParameter() throws Exception {
+        Map<String, String> processedParameters = new HashMap<>();
+        ConnectionFactory cf = new ConnectionFactory() {
+            @Override
+            protected void processUriQueryParameter(String key, String value) {
+                processedParameters.put(key, value);
+            }
+        };
+        cf.setUri("amqp://user:pass@host:10000/vhost?heartbeat=60&key=value");
+        assertThat(processedParameters).hasSize(1).containsEntry("key", "value");
+    }
+
     private void parseSuccess(String uri, String user, String password,
                               String host, int port, String vhost, boolean secured)
         throws URISyntaxException, NoSuchAlgorithmException, KeyManagementException