Merge bug21647 (allow consumers to be informed of queue death)

Author: Simon MacMullen

build.xml

@@ -18,7 +18,7 @@
 
   <path id="test.javac.classpath">
     <path refid="javac.classpath"/>
-<!--    <pathelement path="${junit.home}/junit.jar"/> -->
+<!--    <pathelement path="${junit.home}/junit.jar"/>  -->
   </path>
 
   <path id="test.classpath">

src/com/rabbitmq/client/Connection.java

@@ -17,6 +17,7 @@
 package com.rabbitmq.client;
 
 import java.io.IOException;
+import java.net.InetAddress;
 import java.util.Map;
 
 /**
@@ -46,7 +47,7 @@ public interface Connection extends ShutdownNotifier { // rename to AMQPConnecti
      * Retrieve the host.
      * @return the hostname of the peer we're connected to.
      */
-    String getHost();
+    InetAddress getAddress();
 
     /**
      * Retrieve the port number.

src/com/rabbitmq/client/DefaultSaslConfig.java

@@ -20,38 +20,54 @@
 import javax.security.sasl.Sasl;
 import javax.security.sasl.SaslClient;
 import javax.security.sasl.SaslException;
-import java.util.Map;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
 
 /**
  * Default implementation of SaslConfig that uses the standard Java
  * algorithm for selecting a sasl client.
  * @see com.rabbitmq.client.ConnectionFactory
  */
 public class DefaultSaslConfig implements SaslConfig {
-    private ConnectionFactory factory;
-    private String authorizationId;
-    private Map<String,?> mechanismProperties;
-    private CallbackHandler callbackHandler;
+    public static final String[] DEFAULT_PREFERRED_MECHANISMS = new String[]{"PLAIN"};
 
-    public DefaultSaslConfig(ConnectionFactory factory) {
-        this.factory = factory;
-        callbackHandler = new UsernamePasswordCallbackHandler(factory);
-    }
+    private final ConnectionFactory factory;
+    private final List<String> mechanisms;
+    private final CallbackHandler callbackHandler;
 
-    public void setAuthorizationId(String authorizationId) {
-        this.authorizationId = authorizationId;
+    /**
+     * Create a DefaultSaslConfig which only wants to use PLAIN.
+     *
+     * @param factory - the ConnectionFactory to use to obtain username, password and host
+     */
+    public DefaultSaslConfig(ConnectionFactory factory) {
+        this(factory, DEFAULT_PREFERRED_MECHANISMS);
     }
 
-    public void setMechanismProperties(Map<String, ?> mechanismProperties) {
-        this.mechanismProperties = mechanismProperties;
+    /**
+     * Create a DefaultSaslConfig with a list of mechanisms to use.
+     *
+     * @param factory - the ConnectionFactory to use to obtain username, password and host
+     * @param mechanisms - a list of SASL mechanisms to use (in descending order of preference)
+     */
+    public DefaultSaslConfig(ConnectionFactory factory, String[] mechanisms) {
+        this.factory = factory;
+        callbackHandler = new UsernamePasswordCallbackHandler(factory);
+        this.mechanisms = Arrays.asList(mechanisms);
     }
 
-    public void setCallbackHandler(CallbackHandler callbackHandler) {
-        this.callbackHandler = callbackHandler;
-    }
+    public SaslClient getSaslClient(String[] serverMechanisms) throws SaslException {
+        Set<String> server = new HashSet<String>(Arrays.asList(serverMechanisms));
 
-    public SaslClient getSaslClient(String[] mechanisms) throws SaslException {
-        return Sasl.createSaslClient(mechanisms, authorizationId, "AMQP",
-              factory.getHost(), mechanismProperties, callbackHandler);
+        for (String mechanism: mechanisms) {
+            if (server.contains(mechanism)) {
+                SaslClient saslClient = Sasl.createSaslClient(new String[]{mechanism},
+                         null, "AMQP", factory.getHost(), null, callbackHandler);
+                if (saslClient != null) return saslClient;
+            }
+        }
+        return null;
     }
 }

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

@@ -19,6 +19,7 @@
 
 import java.io.EOFException;
 import java.io.IOException;
+import java.net.InetAddress;
 import java.net.SocketException;
 import java.util.HashMap;
 import java.util.Map;
@@ -152,8 +153,8 @@ public void ensureIsOpen()
     public Map<String, Object> _serverProperties;
 
     /** {@inheritDoc} */
-    public String getHost() {
-        return _frameHandler.getHost();
+    public InetAddress getAddress() {
+        return _frameHandler.getAddress();
     }
 
     /** {@inheritDoc} */
@@ -242,7 +243,7 @@ public void start()
 
         // start the main loop going
         Thread ml = new MainLoop();
-        ml.setName("AMQP Connection " + getHost() + ":" + getPort());
+        ml.setName("AMQP Connection " + getHostAddress() + ":" + getPort());
         ml.start();
 
         AMQP.Connection.Start connStart = null;
@@ -560,7 +561,7 @@ public void handleConnectionClose(Command closeCommand) {
         _brokerInitiatedShutdown = true;
         Thread scw = new SocketCloseWait(sse);
         scw.setName("AMQP Connection Closing Monitor " +
-                    getHost() + ":" + getPort());
+                getHostAddress() + ":" + getPort());
         scw.start();
     }
 
@@ -727,6 +728,10 @@ public void close(int closeCode,
     }
 
     @Override public String toString() {
-        return "amqp://" + _factory.getUsername() + "@" + getHost() + ":" + getPort() + _virtualHost;
+        return "amqp://" + _factory.getUsername() + "@" + getHostAddress() + ":" + getPort() + _virtualHost;
+    }
+
+    private String getHostAddress() {
+        return getAddress() == null ? null : getAddress().getHostAddress();
     }
 }

src/com/rabbitmq/client/impl/FrameHandler.java

@@ -17,6 +17,7 @@
 package com.rabbitmq.client.impl;
 
 import java.io.IOException;
+import java.net.InetAddress;
 import java.net.SocketException;
 import java.net.SocketTimeoutException;
 
@@ -25,8 +26,8 @@
  */
 
 public interface FrameHandler {
-    /** Retrieve hostname of peer. */
-    public String getHost();
+    /** Retrieve address of peer. */
+    public InetAddress getAddress();
 
     /** Retrieve port number of peer. */
     public int getPort();

src/com/rabbitmq/client/impl/SocketFrameHandler.java

@@ -21,6 +21,7 @@
 import java.io.DataInputStream;
 import java.io.DataOutputStream;
 import java.io.IOException;
+import java.net.InetAddress;
 import java.net.Socket;
 import java.net.SocketException;
 
@@ -55,8 +56,8 @@ public SocketFrameHandler(Socket socket) throws IOException {
         _outputStream = new DataOutputStream(new BufferedOutputStream(_socket.getOutputStream()));
     }
 
-    public String getHost() {
-        return _socket.getInetAddress().getHostName();
+    public InetAddress getAddress() {
+        return _socket.getInetAddress();
     }
 
     public int getPort() {

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

@@ -17,6 +17,7 @@
 package com.rabbitmq.client.test;
 
 import java.io.IOException;
+import java.net.InetAddress;
 import java.net.SocketException;
 import java.net.SocketTimeoutException;
 import java.util.ArrayList;
@@ -159,8 +160,8 @@ public int getTimeout() throws SocketException {
             return 0;
         }
 
-        public String getHost() {
-            return "MockFrameHandler";
+        public InetAddress getAddress() {
+            return null;
         }
 
         public int getPort() {

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

@@ -18,6 +18,7 @@
 package com.rabbitmq.client.test;
 
 import java.io.IOException;
+import java.net.InetAddress;
 import java.net.SocketException;
 import java.util.ArrayList;
 import java.util.Iterator;
@@ -145,8 +146,8 @@ public int getTimeout() throws SocketException {
             return 0;
         }
 
-        public String getHost() {
-            return "MyFrameHandler";
+        public InetAddress getAddress() {
+            return null;
         }
 
         public int getPort() {