Merge bug22947 into default.

Author: Simon MacMullen

build.properties

@@ -13,7 +13,7 @@ test.javac.out=${build.out}/test/classes
 test.src.home=test/src
 java-jvm-1.4=c:/Program Files/java/j2re1.4.2_14/bin/java
 sibling.codegen.dir=../rabbitmq-codegen/
-spec.version=0.8
+spec.version=0.9.1
 bundle.out=${build.out}/bundle
 javadoc.out=build/doc/api
 python.bin=python

codegen.py

@@ -150,6 +150,7 @@ def printHeader():
     public static class PROTOCOL {"""
         print "        public static final int MAJOR = %i;" % spec.major
         print "        public static final int MINOR = %i;" % spec.minor
+        print "        public static final int REVISION = %i;" % spec.revision
         print "        public static final int PORT = %i;" % spec.port
         print "    }"
 

src/com/rabbitmq/client/BasicProperties.java

@@ -114,12 +114,6 @@ public interface BasicProperties {
      */
     public abstract String getAppId();
 
-    /**
-     * Retrieve the value in the clusterId field.
-     * @return clusterId field, or null if the field has not been set.
-     */
-    public abstract String getClusterId();
-
     /**
      * Set the contentType field, or null indicating the field is not set
      * @param contentType the value to set the field to
@@ -198,9 +192,4 @@ public interface BasicProperties {
      */
     public abstract void setAppId(String appId);
 
-    /**
-     * Set the clusterId field, or null indicating the field is not set
-     * @param clusterId the value to set the field to
-     */
-    public abstract void setClusterId(String clusterId);
 }

src/com/rabbitmq/client/Channel.java

@@ -478,7 +478,7 @@ Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, b
      * @see com.rabbitmq.client.AMQP.Basic.Consume
      * @see com.rabbitmq.client.AMQP.Basic.ConsumeOk
      * @see #basicAck
-     * @see #basicConsume(String,boolean, String,boolean,boolean, Consumer)
+     * @see #basicConsume(String,boolean, String,boolean,boolean, Map, Consumer)
      */
     String basicConsume(String queue, Consumer callback) throws IOException;
 
@@ -492,7 +492,7 @@ Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, b
      * @throws java.io.IOException if an error is encountered
      * @see com.rabbitmq.client.AMQP.Basic.Consume
      * @see com.rabbitmq.client.AMQP.Basic.ConsumeOk
-     * @see #basicConsume(String,boolean, String,boolean,boolean, Consumer)
+     * @see #basicConsume(String,boolean, String,boolean,boolean, Map, Consumer)
      */
     String basicConsume(String queue, boolean noAck, Consumer callback) throws IOException;
 
@@ -506,7 +506,7 @@ Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, b
      * @throws java.io.IOException if an error is encountered
      * @see com.rabbitmq.client.AMQP.Basic.Consume
      * @see com.rabbitmq.client.AMQP.Basic.ConsumeOk
-     * @see #basicConsume(String,boolean, String,boolean,boolean, Consumer)
+     * @see #basicConsume(String,boolean, String,boolean,boolean, Map, Consumer)
      */
     String basicConsume(String queue, boolean noAck, String consumerTag, Consumer callback) throws IOException;
 
@@ -524,7 +524,7 @@ Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, b
      * @see com.rabbitmq.client.AMQP.Basic.Consume
      * @see com.rabbitmq.client.AMQP.Basic.ConsumeOk
      */
-    String basicConsume(String queue, boolean noAck, String consumerTag, boolean noLocal, boolean exclusive, Consumer callback) throws IOException;
+    String basicConsume(String queue, boolean noAck, String consumerTag, boolean noLocal, boolean exclusive, Map<String, Object> filter, Consumer callback) throws IOException;
 
     /**
      * Cancel a consumer. Calls the consumer's {@link Consumer#handleCancelOk}

src/com/rabbitmq/client/Connection.java

@@ -98,12 +98,6 @@ public interface Connection extends ShutdownNotifier { // rename to AMQPConnecti
      */
     Map<String, Object> getClientProperties();
 
-    /**
-     * Retrieve the known hosts.
-     * @return an array of addresses for all hosts that came back in the initial {@link com.rabbitmq.client.AMQP.Connection.OpenOk} open-ok method
-     */
-    Address[] getKnownHosts();
-
     /**
      * Retrieve the server properties.
      * @return a map of the server properties. This typically includes the product name and version of the server.

src/com/rabbitmq/client/ConnectionFactory.java

@@ -361,53 +361,25 @@ protected void configureSocket(Socket socket) throws IOException{
         socket.setTcpNoDelay(true);
     }
 
-    private Connection newConnection(Address[] addresses,
-                                     int maxRedirects,
-                                     Map<Address,Integer> redirectAttempts)
+    /**
+     * Create a new broker connection
+     * @param addrs an array of known broker addresses (hostname/port pairs) to try in order
+     * @return an interface to the connection
+     * @throws IOException if it encounters a problem
+     */
+    public Connection newConnection(Address[] addrs)
         throws IOException
     {
         IOException lastException = null;
-        for (Address address : addresses) {
-            Address[] lastKnownAddresses = new Address[0];
+        for (Address addr : addrs) {
             try {
-                while(true) {
-                    FrameHandler frameHandler = createFrameHandler(address);
-                    Integer redirectCount = redirectAttempts.get(address);
-                    if (redirectCount == null)
-                        redirectCount = 0;
-                    boolean allowRedirects = redirectCount < maxRedirects;
-                    try {
-                        AMQConnection conn = new AMQConnection(this,
-                                    frameHandler);
-                        conn.start(!allowRedirects);
-                        return conn;
-                    } catch (RedirectException e) {
-                        if (!allowRedirects) {
-                            //this should never happen with a well-behaved server
-                            throw new IOException("server ignored 'insist'");
-                        } else {
-                            redirectAttempts.put(address, redirectCount+1);
-                            lastKnownAddresses = e.getKnownAddresses();
-                            address = e.getAddress();
-                            //TODO: we may want to log redirection attempts.
-                        }
-                    }
-                }
+                FrameHandler frameHandler = createFrameHandler(addr);
+                AMQConnection conn = new AMQConnection(this,
+                                                     frameHandler);
+                conn.start();
+                return conn;
             } catch (IOException e) {
                 lastException = e;
-                if (lastKnownAddresses.length > 0) {
-                    // If there aren't any, don't bother trying, since
-                    // a recursive call with empty lastKnownAddresses
-                    // will cause our lastException to be stomped on
-                    // by an uninformative IOException. See bug 16273.
-                    try {
-                        return newConnection(lastKnownAddresses,
-                                             maxRedirects,
-                                             redirectAttempts);
-                    } catch (IOException e1) {
-                        lastException = e1;
-                    }
-                }
             }
         }
 
@@ -425,10 +397,7 @@ private Connection newConnection(Address[] addresses,
      */
     public Connection newConnection() throws IOException {
         return newConnection(new Address[] {
-                                 new Address(getHost(), getPort())
-                             },
-                             0,
-                             new HashMap<Address,Integer>());
+                                 new Address(getHost(), getPort())});
     }
 
 

src/com/rabbitmq/client/RedirectException.java

@@ -46,7 +46,6 @@
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
 import com.rabbitmq.client.MissedHeartbeatException;
-import com.rabbitmq.client.RedirectException;
 import com.rabbitmq.client.ShutdownSignalException;
 import com.rabbitmq.utility.BlockingCell;
 import com.rabbitmq.utility.Utility;
@@ -155,9 +154,6 @@ public void ensureIsOpen()
      */
     private int _heartbeat;
 
-    /** Hosts retrieved from the connection.open-ok */
-    private Address[] _knownHosts;
-
     private final String _username, _password, _virtualHost;
     private final int _requestedChannelMax, _requestedFrameMax, _requestedHeartbeat;
     private final Map<String, Object> _clientProperties;
@@ -175,11 +171,6 @@ public int getPort() {
         return _frameHandler.getPort();
     }
 
-    /** {@inheritDoc} */
-    public Address[] getKnownHosts() {
-        return _knownHosts;
-    }
-
     public FrameHandler getFrameHandler(){
         return _frameHandler;
     }
@@ -236,12 +227,10 @@ public AMQConnection(ConnectionFactory factory,
      * Connection.Start/.StartOk, Connection.Tune/.TuneOk, and then
      * calls Connection.Open and waits for the OpenOk. Sets heartbeat
      * and frame max values after tuning has taken place.
-     * @param insist true if broker redirects are disallowed
-     * @throws RedirectException if the server is redirecting us to a different host/port
      * @throws java.io.IOException if an error is encountered
      */
-    public void start(boolean insist)
-        throws IOException, RedirectException
+    public void start()
+        throws IOException
     {
         // Make sure that the first thing we do is to send the header,
         // which should cause any socket errors to show up for us, rather
@@ -316,19 +305,11 @@ public void start(boolean insist)
         _channel0.transmit(new AMQImpl.Connection.TuneOk(channelMax,
                                                          frameMax,
                                                          heartbeat));
-        
+        // 0.9.1: insist [on not being redirected] is deprecated, but
+        // still in generated code; just pass a dummy value here
         Method res = _channel0.exnWrappingRpc(new AMQImpl.Connection.Open(_virtualHost,
                                                                           "",
-                                                                          insist)).getMethod();
-        if (res instanceof AMQP.Connection.Redirect) {
-            AMQP.Connection.Redirect redirect = (AMQP.Connection.Redirect) res;
-            throw new RedirectException(Address.parseAddress(redirect.getHost()),
-                                        Address.parseAddresses(redirect.getKnownHosts()));
-        } else {
-            AMQP.Connection.OpenOk openOk = (AMQP.Connection.OpenOk) res;
-            _knownHosts = Address.parseAddresses(openOk.getKnownHosts());
-        }
-        
+                                                                          false)).getMethod();
         return;
     }
 

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

@@ -76,7 +76,7 @@ public class ChannelN extends AMQChannel implements com.rabbitmq.client.Channel
      * and this field can be deleted.
      */
     @Deprecated
-    private static final int TICKET = 1;
+    private static final int TICKET = 0;
 
     /**
      * Map from consumer tag to {@link Consumer} instance.
@@ -672,12 +672,12 @@ public String basicConsume(String queue, boolean noAck, String consumerTag,
                                Consumer callback)
         throws IOException
     {
-        return basicConsume(queue, noAck, consumerTag, false, false, callback);
+        return basicConsume(queue, noAck, consumerTag, false, false, null, callback);
     }
 
     /** Public API - {@inheritDoc} */
     public String basicConsume(String queue, boolean noAck, String consumerTag,
-                               boolean noLocal, boolean exclusive,
+                               boolean noLocal, boolean exclusive, Map<String, Object> filter,
                                final Consumer callback)
         throws IOException
     {
@@ -702,7 +702,7 @@ public String transformReply(AMQCommand replyCommand) {
 
         rpc(new Basic.Consume(TICKET, queue, consumerTag,
                               noLocal, noAck, exclusive,
-                              false),
+                              false, filter),
             k);
 
         try {

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

@@ -173,12 +173,32 @@ public static void protocolVersionMismatch(DataInputStream is) throws IOExceptio
         }
 
         try {
-            int transportHigh = is.readUnsignedByte();
-            int transportLow = is.readUnsignedByte();
-            int serverMajor = is.readUnsignedByte();
-            int serverMinor = is.readUnsignedByte();
-            x = new MalformedFrameException("AMQP protocol version mismatch; we are version " + AMQP.PROTOCOL.MAJOR + "-" + AMQP.PROTOCOL.MINOR
-                    + ", server is " + serverMajor + "-" + serverMinor + " with transport " + transportHigh + "." + transportLow);
+            int[] signature = new int[4];
+
+            for (int i = 0; i < 4; i++) {
+                signature[i] = is.readUnsignedByte();
+            }
+
+            if (signature[0] == 1 &&
+                signature[1] == 1 &&
+                signature[2] == 8 &&
+                signature[3] == 0) {
+                x = new MalformedFrameException("AMQP protocol version mismatch; we are version " +
+                        AMQP.PROTOCOL.MAJOR + "-" + AMQP.PROTOCOL.MINOR + "-" + AMQP.PROTOCOL.REVISION +
+                        ", server is 0-8");
+            }
+            else {
+                String sig = "";
+                for (int i = 0; i < 4; i++) {
+                    if (i != 0) sig += ",";
+                    sig += signature[i];
+                }
+
+                x = new MalformedFrameException("AMQP protocol version mismatch; we are version " +
+                        AMQP.PROTOCOL.MAJOR + "-" + AMQP.PROTOCOL.MINOR + "-" + AMQP.PROTOCOL.REVISION +
+                        ", server sent signature " + sig);
+            }
+
         } catch (IOException ex) {
             x = new MalformedFrameException("Invalid AMQP protocol header from server");
         }