Merge branch '4.x.x-stable' into 5.1.x-stable

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

Author: acogoluegnes

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

@@ -15,25 +15,41 @@
 
 package com.rabbitmq.client;
 
-import com.rabbitmq.client.impl.*;
+import static java.util.concurrent.TimeUnit.*;
+
+import com.rabbitmq.client.impl.AMQConnection;
+import com.rabbitmq.client.impl.ConnectionParams;
+import com.rabbitmq.client.impl.CredentialsProvider;
+import com.rabbitmq.client.impl.DefaultCredentialsProvider;
+import com.rabbitmq.client.impl.DefaultExceptionHandler;
+import com.rabbitmq.client.impl.ErrorOnWriteListener;
+import com.rabbitmq.client.impl.FrameHandler;
+import com.rabbitmq.client.impl.FrameHandlerFactory;
+import com.rabbitmq.client.impl.SocketFrameHandlerFactory;
 import com.rabbitmq.client.impl.nio.NioParams;
 import com.rabbitmq.client.impl.nio.SocketChannelFrameHandlerFactory;
 import com.rabbitmq.client.impl.recovery.AutorecoveringConnection;
-
-import javax.net.SocketFactory;
-import javax.net.ssl.SSLContext;
-import javax.net.ssl.SSLSocketFactory;
-import javax.net.ssl.TrustManager;
 import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.net.URLDecoder;
 import java.security.KeyManagementException;
 import java.security.NoSuchAlgorithmException;
-import java.util.*;
-import java.util.concurrent.*;
-
-import static java.util.concurrent.TimeUnit.*;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeoutException;
+import javax.net.SocketFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocketFactory;
+import javax.net.ssl.TrustManager;
 
 /**
  * Convenience factory class to facilitate opening a {@link Connection} to a RabbitMQ node.
@@ -42,7 +58,6 @@
  * Some settings that apply to connections can also be configured here
  * and will apply to all connections produced by this factory.
  */
-
 public class ConnectionFactory implements Cloneable {
 
     /** Default user name */
@@ -92,8 +107,6 @@ public class ConnectionFactory implements Cloneable {
 
     private static final String FALLBACK_TLS_PROTOCOL = "TLSv1";
 
-    private String username                       = DEFAULT_USER;
-    private String password                       = DEFAULT_PASS;
     private String virtualHost                    = DEFAULT_VHOST;
     private String host                           = DEFAULT_HOST;
     private int port                              = USE_DEFAULT_PORT;
@@ -106,17 +119,19 @@ public class ConnectionFactory implements Cloneable {
     private Map<String, Object> _clientProperties = AMQConnection.defaultClientProperties();
     private SocketFactory socketFactory           = null;
     private SaslConfig saslConfig                 = DefaultSaslConfig.PLAIN;
+
     private ExecutorService sharedExecutor;
-    private ThreadFactory threadFactory           = Executors.defaultThreadFactory();
+    private ThreadFactory threadFactory             = Executors.defaultThreadFactory();
     // minimises the number of threads rapid closure of many
     // connections uses, see rabbitmq/rabbitmq-java-client#86
     private ExecutorService shutdownExecutor;
     private ScheduledExecutorService heartbeatExecutor;
-    private SocketConfigurator socketConf         = new DefaultSocketConfigurator();
-    private ExceptionHandler exceptionHandler     = new DefaultExceptionHandler();
+    private SocketConfigurator socketConf           = new DefaultSocketConfigurator();
+    private ExceptionHandler exceptionHandler       = new DefaultExceptionHandler();
+    private CredentialsProvider credentialsProvider = new DefaultCredentialsProvider(DEFAULT_USER, DEFAULT_PASS);
 
-    private boolean automaticRecovery             = true;
-    private boolean topologyRecovery              = true;
+    private boolean automaticRecovery               = true;
+    private boolean topologyRecovery                = true;
 
     // long is used to make sure the users can use both ints
     // and longs safely. It is unlikely that anybody'd need
@@ -193,33 +208,50 @@ public void setPort(int port) {
      * @return the AMQP user name to use when connecting to the broker
      */
     public String getUsername() {
-        return this.username;
+        return credentialsProvider.getUsername();
     }
 
     /**
      * Set the user name.
      * @param username the AMQP user name to use when connecting to the broker
      */
     public void setUsername(String username) {
-        this.username = username;
+        this.credentialsProvider = new DefaultCredentialsProvider(
+            username,
+            this.credentialsProvider.getPassword()
+        );
     }
 
     /**
      * Retrieve the password.
      * @return the password to use when connecting to the broker
      */
     public String getPassword() {
-        return this.password;
+        return credentialsProvider.getPassword();
     }
 
     /**
      * Set the password.
      * @param password the password to use when connecting to the broker
      */
     public void setPassword(String password) {
-        this.password = password;
+        this.credentialsProvider = new DefaultCredentialsProvider(
+            this.credentialsProvider.getUsername(),
+            password
+        );
     }
 
+    /**
+     * Set a custom credentials provider.
+     * Default implementation uses static username and password.
+     * @param credentialsProvider The custom implementation of CredentialsProvider to use when connecting to the broker.
+     * @see com.rabbitmq.client.impl.DefaultCredentialsProvider
+     * @since 4.5.0
+     */
+    public void setCredentialsProvider(CredentialsProvider credentialsProvider) {
+        this.credentialsProvider = credentialsProvider;
+    }
+    
     /**
      * Retrieve the virtual host.
      * @return the virtual host to use when connecting to the broker
@@ -994,8 +1026,7 @@ public Connection newConnection(ExecutorService executor, AddressResolver addres
     public ConnectionParams params(ExecutorService consumerWorkServiceExecutor) {
         ConnectionParams result = new ConnectionParams();
 
-        result.setUsername(username);
-        result.setPassword(password);
+        result.setCredentialsProvider(credentialsProvider);
         result.setConsumerWorkServiceExecutor(consumerWorkServiceExecutor);
         result.setVirtualHost(virtualHost);
         result.setClientProperties(getClientProperties());
@@ -1095,7 +1126,8 @@ protected AddressResolver createAddressResolver(List<Address> addresses) {
 
     @Override public ConnectionFactory clone(){
         try {
-            return (ConnectionFactory)super.clone();
+            ConnectionFactory clone = (ConnectionFactory)super.clone();
+            return clone;
         } catch (CloneNotSupportedException e) {
             throw new RuntimeException(e);
         }
@@ -1331,7 +1363,7 @@ public boolean isChannelShouldCheckRpcResponseType() {
 
     /**
      * Timeout (in ms) for work pool enqueueing.
-     * The {@link WorkPool} dispatches several types of responses
+     * The {@link com.rabbitmq.client.impl.WorkPool} dispatches several types of responses
      * from the broker (e.g. deliveries). A high-traffic
      * client with slow consumers can exhaust the work pool and
      * compromise the whole connection (by e.g. letting the broker

src/main/java/com/rabbitmq/client/impl/AMQConnection.java

@@ -136,8 +136,7 @@ public static Map<String, Object> defaultClientProperties() {
     private final int requestedFrameMax;
     private final int handshakeTimeout;
     private final int shutdownTimeout;
-    private final String username;
-    private final String password;
+    private final CredentialsProvider credentialsProvider;
     private final Collection<BlockedListener> blockedListeners = new CopyOnWriteArrayList<BlockedListener>();
     protected final MetricsCollector metricsCollector;
     private final int channelRpcTimeout;
@@ -216,8 +215,7 @@ public AMQConnection(ConnectionParams params, FrameHandler frameHandler) {
     public AMQConnection(ConnectionParams params, FrameHandler frameHandler, MetricsCollector metricsCollector)
     {
         checkPreconditions();
-        this.username = params.getUsername();
-        this.password = params.getPassword();
+        this.credentialsProvider = params.getCredentialsProvider();
         this._frameHandler = frameHandler;
         this._virtualHost = params.getVirtualHost();
         this._exceptionHandler = params.getExceptionHandler();
@@ -337,8 +335,10 @@ public void start()
                                               "server offered [" + connStart.getMechanisms() + "]");
             }
 
+            String username = credentialsProvider.getUsername();
+            String password = credentialsProvider.getPassword();
             LongString challenge = null;
-            LongString response = sm.handleChallenge(null, this.username, this.password);
+            LongString response = sm.handleChallenge(null, username, password);
 
             do {
                 Method method = (challenge == null)
@@ -355,7 +355,7 @@ public void start()
                         connTune = (AMQP.Connection.Tune) serverResponse;
                     } else {
                         challenge = ((AMQP.Connection.Secure) serverResponse).getChallenge();
-                        response = sm.handleChallenge(challenge, this.username, this.password);
+                        response = sm.handleChallenge(challenge, username, password);
                     }
                 } catch (ShutdownSignalException e) {
                     Method shutdownMethod = e.getReason();
@@ -1069,7 +1069,7 @@ public AMQCommand transformReply(AMQCommand command) {
 
     @Override public String toString() {
         final String virtualHost = "/".equals(_virtualHost) ? _virtualHost : "/" + _virtualHost;
-        return "amqp://" + this.username + "@" + getHostAddress() + ":" + getPort() + virtualHost;
+        return "amqp://" + this.credentialsProvider.getUsername() + "@" + getHostAddress() + ":" + getPort() + virtualHost;
     }
 
     private String getHostAddress() {

src/main/java/com/rabbitmq/client/impl/ConnectionParams.java

@@ -26,8 +26,7 @@
 import java.util.concurrent.ThreadFactory;
 
 public class ConnectionParams {
-    private String username;
-    private String password;
+    private CredentialsProvider credentialsProvider;
     private ExecutorService consumerWorkServiceExecutor;
     private ScheduledExecutorService heartbeatExecutor;
     private ExecutorService shutdownExecutor;
@@ -52,12 +51,8 @@ public class ConnectionParams {
 
     public ConnectionParams() {}
 
-    public String getUsername() {
-        return username;
-    }
-
-    public String getPassword() {
-        return password;
+    public CredentialsProvider getCredentialsProvider() {
+        return credentialsProvider;
     }
 
     public ExecutorService getConsumerWorkServiceExecutor() {
@@ -132,12 +127,8 @@ public boolean channelShouldCheckRpcResponseType() {
         return channelShouldCheckRpcResponseType;
     }
 
-    public void setUsername(String username) {
-        this.username = username;
-    }
-
-    public void setPassword(String password) {
-        this.password = password;
+    public void setCredentialsProvider(CredentialsProvider credentialsProvider) {
+        this.credentialsProvider = credentialsProvider;
     }
 
     public void setConsumerWorkServiceExecutor(ExecutorService consumerWorkServiceExecutor) {

src/main/java/com/rabbitmq/client/impl/CredentialsProvider.java

@@ -0,0 +1,16 @@
+package com.rabbitmq.client.impl;
+
+/**
+ * Provider interface for establishing credentials for connecting to the broker. Especially useful
+ * for situations where credentials might change before a recovery takes place or where it is 
+ * convenient to plug in an outside custom implementation.
+ *
+ * @since 4.5.0
+ */
+public interface CredentialsProvider {
+
+    String getUsername();
+
+    String getPassword();
+
+}

src/main/java/com/rabbitmq/client/impl/DefaultCredentialsProvider.java

@@ -0,0 +1,29 @@
+package com.rabbitmq.client.impl;
+
+/**
+ * Default implementation of a CredentialsProvider which simply holds a static
+ * username and password.
+ *
+ * @since 4.5.0
+ */
+public class DefaultCredentialsProvider implements CredentialsProvider {
+
+    private final String username;
+    private final String password;
+
+    public DefaultCredentialsProvider(String username, String password) {
+        this.username = username;
+        this.password = password;
+    }
+
+    @Override
+    public String getUsername() {
+        return username;
+    }
+
+    @Override
+    public String getPassword() {
+        return password;
+    }
+    
+}

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

@@ -21,6 +21,7 @@
 import com.rabbitmq.client.MetricsCollector;
 import com.rabbitmq.client.impl.AMQConnection;
 import com.rabbitmq.client.impl.ConnectionParams;
+import com.rabbitmq.client.impl.CredentialsProvider;
 import com.rabbitmq.client.impl.FrameHandler;
 import com.rabbitmq.client.impl.FrameHandlerFactory;
 import org.junit.Test;
@@ -29,8 +30,9 @@
 import java.util.Queue;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
 
-import static org.junit.Assert.assertSame;
+import static org.junit.Assert.*;
 import static org.mockito.Mockito.*;
 
 public class ConnectionFactoryTest {
@@ -62,5 +64,30 @@ protected synchronized FrameHandlerFactory createFrameHandlerFactory() throws IO
         );
         assertSame(connectionThatSucceeds, returnedConnection);
     }
+    
+    // see https://github.com/rabbitmq/rabbitmq-java-client/pull/350
+    @Test public void customizeCredentialsProvider() throws Exception {
+        final CredentialsProvider provider = mock(CredentialsProvider.class);
+        final AMQConnection connection = mock(AMQConnection.class);
+        final AtomicBoolean createCalled = new AtomicBoolean(false);
+        
+        ConnectionFactory connectionFactory = new ConnectionFactory() {
+            @Override
+            protected AMQConnection createConnection(ConnectionParams params, FrameHandler frameHandler,
+                    MetricsCollector metricsCollector) {
+                assertSame(provider, params.getCredentialsProvider());
+                createCalled.set(true);
+                return connection;
+            }
+        };
+        connectionFactory.setCredentialsProvider(provider);
+        connectionFactory.setAutomaticRecoveryEnabled(false);
+        
+        doNothing().when(connection).start();
+        
+        Connection returnedConnection = connectionFactory.newConnection();
+        assertSame(returnedConnection, connection);
+        assertTrue(createCalled.get());
+    }
 
 }