Merge with upstream

Author: essiene

src/com/rabbitmq/client/Connection.java

@@ -75,12 +75,13 @@ public interface Connection extends ShutdownNotifier { // rename to AMQPConnecti
     ConnectionParameters getParameters();
 
     /**
-     * Get the negotiated maximum number of channels allowed.
+     * Get the negotiated maximum channel number. Usable channel
+     * numbers range from 1 to this number, inclusive.
      *
      * Note that this is the <i>current</i> setting, as opposed to the <i>initially-requested</i>
      * setting available from {@link #getParameters()}.{@link ConnectionParameters#getRequestedChannelMax()}.
      *
-     * @return the maximum number of simultaneously-open channels permitted for this connection.
+     * @return the maximum channel number permitted for this connection.
      */
     int getChannelMax();
 

src/com/rabbitmq/client/ConnectionParameters.java

@@ -54,7 +54,7 @@ private static String safeGetProperty(String key, String def) {
     /** Default virtual host */
     public static final String DEFAULT_VHOST = "/";
 
-    /** Default value for the desired maximum number of channels; zero for
+    /** Default value for the desired maximum channel number; zero for
      * unlimited */
     public static final int DEFAULT_CHANNEL_MAX = 0;
 
@@ -126,8 +126,8 @@ public void setVirtualHost(String virtualHost) {
     }
 
     /**
-     * Retrieve the requested maximum number of channels
-     * @return the initially requested maximum number of channels; zero for unlimited
+     * Retrieve the requested maximum channel number
+     * @return the initially requested maximum channel number; zero for unlimited
      */
     public int getRequestedChannelMax() {
         return _requestedChannelMax;
@@ -166,8 +166,8 @@ public void setRequestedHeartbeat(int requestedHeartbeat) {
     }
 
     /**
-     * Set the requested maximum number of channels
-     * @param requestedChannelMax initially requested maximum number of channels; zero for unlimited
+     * Set the requested maximum channel number
+     * @param requestedChannelMax initially requested maximum channel number; zero for unlimited
      */
     public void setRequestedChannelMax(int requestedChannelMax) {
         _requestedChannelMax = requestedChannelMax;

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

@@ -219,7 +219,7 @@ public int getChannelMax() {
     }
 
     /**
-     * Protected API - set the max <b>number</b> of channels available
+     * Protected API - set the max channel <b>number</b>
      */
     public void setChannelMax(int value) {
         _channelManager.setChannelMax(value);

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

@@ -47,7 +47,7 @@ public class ChannelManager {
     /** Mapping from channel number to AMQChannel instance */
     private final Map<Integer, ChannelN> _channelMap = Collections.synchronizedMap(new HashMap<Integer, ChannelN>());
 
-    /** Maximum number of channels available on this connection. */
+    /** Maximum channel number available on this connection. */
     public int _channelMax = 0;
 
     public synchronized int getChannelMax() {
@@ -101,7 +101,7 @@ public synchronized int allocateChannelNumber(int maxChannels) {
             maxChannels = Integer.MAX_VALUE;
         }
         int channelNumber = -1;
-        for (int candidate = 1; candidate < maxChannels; candidate++) {
+        for (int candidate = 1; candidate <= maxChannels; candidate++) {
             if (!_channelMap.containsKey(candidate)) {
                 channelNumber = candidate;
                 break;

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

@@ -193,8 +193,12 @@ public void releaseChannelNumber() {
                                                                          command,
                                                                          this);
             synchronized (_channelMutex) {
-                processShutdownSignal(signal, true, true);
-                quiescingTransmit(new Channel.CloseOk());
+                try {
+                    processShutdownSignal(signal, true, false);
+                    quiescingTransmit(new Channel.CloseOk());
+                } finally {
+                    notifyOutstandingRpc(signal);
+                }
             }
             notifyListeners();
             return true;

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

@@ -36,7 +36,7 @@
 
 /**
  * An object providing access to a LongString.
- * This might be implemeted to read directly from connection
+ * This might be implemented to read directly from connection
  * socket, depending on the size of the content to be read -
  * long strings may contain up to 4Gb of content.
  */

test/src/com/rabbitmq/client/test/functional/BindingLifecycle.java

@@ -31,6 +31,8 @@
 
 package com.rabbitmq.client.test.functional;
 
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.GetResponse;
 import com.rabbitmq.client.QueueingConsumer;
 
@@ -53,6 +55,29 @@ public class BindingLifecycle extends PersisterRestartBase {
     protected static final String X = "X-" + System.currentTimeMillis();
     protected static final String K = "K-" + System.currentTimeMillis();
 
+    /**
+     * Create a durable queue on secondary node, if possible, falling
+     * back on the primary node if necessary.
+     */
+    @Override protected void declareDurableQueue(String q)
+        throws IOException
+    {
+        Connection connection;
+        try {
+            connection = connectionFactory.newConnection("localhost", 5673);
+        } catch (IOException e) {
+            super.declareDurableQueue(q);
+            return;
+        }
+
+        Channel channel = connection.createChannel();
+
+        channel.queueDeclare(q, true);
+
+        channel.abort();
+        connection.abort();
+    }
+
     /**
      *   Tests whether durable bindings are correctly recovered.
      */
@@ -112,7 +137,7 @@ public void testDurableBindingsDeletion() throws IOException {
 
 
     /**
-     * This tests whether the default bindings for persistent queues
+     * This tests whether the default bindings for durable queues
      * are recovered properly.
      *
      * The idea is to create a durable queue, nuke the server and then

test/src/com/rabbitmq/client/test/functional/DoubleDeletion.java

@@ -0,0 +1,70 @@
+//   The contents of this file are subject to the Mozilla Public License
+//   Version 1.1 (the "License"); you may not use this file except in
+//   compliance with the License. You may obtain a copy of the License at
+//   http://www.mozilla.org/MPL/
+//
+//   Software distributed under the License is distributed on an "AS IS"
+//   basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
+//   License for the specific language governing rights and limitations
+//   under the License.
+//
+//   The Original Code is RabbitMQ.
+//
+//   The Initial Developers of the Original Code are LShift Ltd,
+//   Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd.
+//
+//   Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd,
+//   Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd
+//   are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial
+//   Technologies LLC, and Rabbit Technologies Ltd.
+//
+//   Portions created by LShift Ltd are Copyright (C) 2007-2009 LShift
+//   Ltd. Portions created by Cohesive Financial Technologies LLC are
+//   Copyright (C) 2007-2009 Cohesive Financial Technologies
+//   LLC. Portions created by Rabbit Technologies Ltd are Copyright
+//   (C) 2007-2009 Rabbit Technologies Ltd.
+//
+//   All Rights Reserved.
+//
+//   Contributor(s): ______________________________________.
+//
+
+package com.rabbitmq.client.test.functional;
+
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.ShutdownSignalException;
+import java.io.IOException;
+
+public class DoubleDeletion extends BrokerTestCase
+{
+    protected static final String Q = "DoubleDeletionQueue";
+    protected static final String X = "DoubleDeletionExchange";
+
+    public void testDoubleDeletionQueue()
+        throws IOException
+    {
+        channel.queueDeclare(Q);
+	channel.queueDelete(Q);
+        try {
+            channel.queueDelete(Q);
+            fail("Expected exception from double deletion of queue");
+        } catch (IOException ee) {
+	    checkShutdownSignal(AMQP.NOT_FOUND, ee);
+            // Pass!
+        }
+    }
+
+    public void testDoubleDeletionExchange()
+        throws IOException
+    {
+        channel.exchangeDeclare(X, "direct");
+	channel.exchangeDelete(X);
+        try {
+            channel.exchangeDelete(X);
+            fail("Expected exception from double deletion of exchange");
+        } catch (IOException ee) {
+	    checkShutdownSignal(AMQP.NOT_FOUND, ee);
+            // Pass!
+        }
+    }
+}

test/src/com/rabbitmq/client/test/functional/FunctionalTests.java

@@ -37,6 +37,7 @@
 public class FunctionalTests extends TestCase {
     public static TestSuite suite() {
         TestSuite suite = new TestSuite("functional");
+        suite.addTestSuite(DoubleDeletion.class);
         suite.addTestSuite(Routing.class);
         suite.addTestSuite(BindingLifecycle.class);
         suite.addTestSuite(Transactions.class);

test/src/com/rabbitmq/client/test/functional/PersisterRestartBase.java

@@ -89,7 +89,7 @@ protected void declareDurableQueue(String q)
     protected void declareAndBindDurableQueue(String q, String x, String r)
         throws IOException
     {
-        channel.queueDeclare(q, true);
+        declareDurableQueue(q);
         channel.queueBind(q, x, r);
     }