Get credentials from a CredentialsProvider, with a default

implementation that simply tracks a static username and password. Useful
for reconnect scenarios where the username or password might have
changed between the connect & reconnect.

Author: Nick Johnson

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

@@ -15,25 +15,40 @@
 
 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.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 an AMQP broker.
@@ -85,8 +100,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;
@@ -107,6 +120,11 @@ public class ConnectionFactory implements Cloneable {
     private ScheduledExecutorService heartbeatExecutor;
     private SocketConfigurator socketConf         = new DefaultSocketConfigurator();
     private ExceptionHandler exceptionHandler     = new DefaultExceptionHandler();
+    private CredentialsProvider credentialsProv   = new DefaultCredentialsProvider();
+    {
+        credentialsProv.setUsername(DEFAULT_USER);
+        credentialsProv.setPassword(DEFAULT_PASS);
+    }
 
     private boolean automaticRecovery             = true;
     private boolean topologyRecovery              = true;
@@ -172,33 +190,43 @@ public void setPort(int port) {
      * @return the AMQP user name to use when connecting to the broker
      */
     public String getUsername() {
-        return this.username;
+        return credentialsProv.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;
+        credentialsProv.setUsername(username);
     }
 
     /**
      * Retrieve the password.
      * @return the password to use when connecting to the broker
      */
     public String getPassword() {
-        return this.password;
+        return credentialsProv.getPassword();
     }
 
     /**
      * Set the password.
      * @param password the password to use when connecting to the broker
      */
     public void setPassword(String password) {
-        this.password = password;
+        credentialsProv.setPassword(password);
     }
 
+    /**
+     * Set a custom credentials provider.
+     * @param credentialsProvider The custom implementation of CredentialsProvider to use when connecting to the broker.
+     * @see com.rabbitmq.client.impl.DefaultCredentialsProvider
+     * @see com.rabbitmq.client.impl.AbstractCredentialsProvider
+     */
+    public void setCredentialsProvider(CredentialsProvider credentialsProvider) {
+        this.credentialsProv = credentialsProvider;
+    }
+    
     /**
      * Retrieve the virtual host.
      * @return the virtual host to use when connecting to the broker
@@ -954,8 +982,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(credentialsProv);
         result.setConsumerWorkServiceExecutor(consumerWorkServiceExecutor);
         result.setVirtualHost(virtualHost);
         result.setClientProperties(getClientProperties());

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

@@ -129,8 +129,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;
@@ -209,8 +208,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();
@@ -323,8 +321,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)
@@ -341,7 +341,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();
@@ -1021,7 +1021,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://" + credentialsProvider.getUsername() + "@" + getHostAddress() + ":" + getPort() + virtualHost;
     }
 
     private String getHostAddress() {

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

@@ -0,0 +1,39 @@
+package com.rabbitmq.client.impl;
+
+/**
+ * Base class for extending to implement a concrete CredentialsProvider, used
+ * when creating connections to the broker or reconnecting during recovery.
+ */
+public abstract class AbstractCredentialsProvider implements CredentialsProvider {
+
+    /* (non-Javadoc)
+     * @see com.rabbitmq.client.impl.CredentialsProvider#getUsername()
+     */
+    @Override
+    public abstract String getUsername();
+
+    /* (non-Javadoc)
+     * @see com.rabbitmq.client.impl.CredentialsProvider#getPassword()
+     */
+    @Override
+    public abstract String getPassword();
+
+    /** 
+     * Default implementation throws UnsupportedOperationException() but can be overridden for 
+     * classes which support setting a static username.
+     */
+    @Override
+    public void setUsername(String username) {
+        throw new UnsupportedOperationException();
+    }
+
+    /** 
+     * Default implementation throws UnsupportedOperationException() but can be overridden for
+     * classes which support setting a static password.
+     */
+    @Override
+    public void setPassword(String password) {
+        throw new UnsupportedOperationException();
+    }
+
+}

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;
@@ -50,12 +49,8 @@ public class ConnectionParams {
 
     public ConnectionParams() {}
 
-    public String getUsername() {
-        return username;
-    }
-
-    public String getPassword() {
-        return password;
+    public CredentialsProvider getCredentialsProvider() {
+        return credentialsProvider;
     }
 
     public ExecutorService getConsumerWorkServiceExecutor() {
@@ -130,12 +125,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,18 @@
+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.
+ */
+public interface CredentialsProvider {
+
+    String getUsername();
+
+    String getPassword();
+
+    void setUsername(String username);
+
+    void setPassword(String password);
+
+}

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

@@ -0,0 +1,32 @@
+package com.rabbitmq.client.impl;
+
+/**
+ * Default implementation of a CredentialsProvider which simply holds a static
+ * username and password.
+ */
+public class DefaultCredentialsProvider extends AbstractCredentialsProvider {
+
+    protected String username;
+    protected String password;
+    
+    @Override
+    public String getUsername() {
+        return username;
+    }
+
+    @Override
+    public String getPassword() {
+        return password;
+    }
+    
+    @Override
+    public void setUsername(String username) {
+        this.username = username;
+    }
+    
+    @Override
+    public void setPassword(String password) {
+        this.password = password;
+    }
+
+}