merges default into bug26129

Author: Alvaro Videla

LICENSE

@@ -1,11 +1,12 @@
-This package, the RabbitMQ Java client library is dual-licensed under
-the MPL and the GPL v2. For the MPL, please see
-LICENSE-MPL-RabbitMQ. For the GPL v2, please see LICENSE-GPL2.
+This package, the RabbitMQ Java client library, is triple-licensed under
+the Mozilla Public License 1.1 ("MPL"), the GNU General Public License
+version 2 ("GPL") and the Apache License version 2 ("ASL"). For the MPL,
+please see LICENSE-MPL-RabbitMQ. For the GPL, please see LICENSE-GPL2.
+For the ASL, please see LICENSE-APACHE2.
 
-The RabbitMQ Java client library includes software under the Apache
-Licence v2. For this license, please see LICENSE-APACHE2. For
-attribution of copyright and other details of provenance, please refer
-to the source code.
+The RabbitMQ Java client library includes third-party software under the
+ASL. For this license, please see LICENSE-APACHE2. For attribution of
+copyright and other details of provenance, please refer to the source code.
 
 If you have any questions regarding licensing, please contact us at 
 info@rabbitmq.com.

LICENSE-MPL-RabbitMQ

@@ -429,7 +429,7 @@
      Initial Developer may designate portions of the Covered Code as
      "Multiple-Licensed".  "Multiple-Licensed" means that the Initial
      Developer permits you to utilize portions of the Covered Code under
-     Your choice of the NPL or the alternative licenses, if any, specified
+     Your choice of the MPL or the alternative licenses, if any, specified
      by the Initial Developer in the file described in Exhibit A.
 
 EXHIBIT A -Mozilla Public License.
@@ -447,8 +447,20 @@ EXHIBIT A -Mozilla Public License.
      The Original Code is RabbitMQ.
 
      The Initial Developer of the Original Code is GoPivotal, Inc.
-     Copyright (c) 2007-2014 GoPivotal, Inc.  All rights reserved.''
-
+     Copyright (c) 2007-2014 GoPivotal, Inc.  All rights reserved.
+
+     Alternatively, the contents of this file may be used under the terms
+     of the GNU General Public License version 2 (the  "GPL2"), or
+     the Apache License version 2 (the "ASL2") in which case the
+     provisions of GPL2 or the ASL2 are applicable instead of those
+     above.  If you wish to allow use of your version of this file only
+     under the terms of the GPL2 or the ASL2 and not to allow others to use
+     your version of this file under the MPL, indicate your decision by
+     deleting  the provisions above and replace  them with the notice and
+     other provisions required by the GPL2 or the ASL2.  If you do not delete
+     the provisions above, a recipient may use your version of this file
+     under either the MPL, the GPL2 or the ASL2.''
+     
      [NOTE: The text of this Exhibit A may differ slightly from the text of
      the notices in the Source Code files of the Original Code. You should
      use the text of this Exhibit A rather than the text found in the

README.in

@@ -5,6 +5,8 @@ For your convenience, a text copy of these instructions is available
 below. Please be aware that the instructions here may not be as up to 
 date as those at the above URL.
 
+See LICENSE for license information.
+
 ===========================================================================
 
 

build.xml

@@ -444,7 +444,7 @@
         </jar>
     </target>
 
-  <target name="dist" depends="jar, test-jar">
+  <target name="dist" depends="jar, test-jar" description="Build all library JARs and documentation">
     <mkdir dir="${dist.out}"/>
     <copy todir="${dist.out}">
       <!-- ant doesn't seem to provide any form of usable abstraction over sets of file names -->
@@ -464,7 +464,7 @@
     </copy>
   </target>
 
-  <target name="clean">
+  <target name="clean" description="Cleans build artifacts">
     <delete dir="build"/>
   </target>
 

src/com/rabbitmq/client/ConnectionFactory.java

@@ -99,7 +99,10 @@ public class ConnectionFactory implements Cloneable {
     private boolean automaticRecovery             = false;
     private boolean topologyRecovery              = true;
 
-    private int networkRecoveryInterval           = 5000;
+    // long is used to make sure the users can use both ints
+    // and longs safely. It is unlikely that anybody'd need
+    // to use recovery intervals > Integer.MAX_VALUE in practice.
+    private long networkRecoveryInterval          = 5000;
 
     /** @return number of consumer threads in default {@link ExecutorService} */
     @Deprecated
@@ -638,7 +641,7 @@ public Connection newConnection(ExecutorService executor) throws IOException {
      * Returns automatic connection recovery interval in milliseconds.
      * @return how long will automatic recovery wait before attempting to reconnect, in ms; default is 5000
      */
-    public int getNetworkRecoveryInterval() {
+    public long getNetworkRecoveryInterval() {
         return networkRecoveryInterval;
     }
 
@@ -649,4 +652,12 @@ public int getNetworkRecoveryInterval() {
     public void setNetworkRecoveryInterval(int networkRecoveryInterval) {
         this.networkRecoveryInterval = networkRecoveryInterval;
     }
+
+    /**
+     * Sets connection recovery interval. Default is 5000.
+     * @param networkRecoveryInterval how long will automatic recovery wait before attempting to reconnect, in ms
+     */
+    public void setNetworkRecoveryInterval(long networkRecoveryInterval) {
+        this.networkRecoveryInterval = networkRecoveryInterval;
+    }
 }

src/com/rabbitmq/client/RpcClient.java

@@ -36,10 +36,10 @@
 import com.rabbitmq.utility.BlockingCell;
 
 /**
- * Convenience class which manages a temporary reply queue for simple RPC-style communication.
+ * Convenience class which manages simple RPC-style communication.
  * The class is agnostic about the format of RPC arguments / return values.
  * It simply provides a mechanism for sending a message to an exchange with a given routing key,
- * and waiting for a response on a reply queue.
+ * and waiting for a response.
 */
 public class RpcClient {
     /** Channel we are communicating on */
@@ -58,8 +58,6 @@ public class RpcClient {
     /** Contains the most recently-used request correlation ID */
     private int _correlationId;
 
-    /** The name of our private reply queue */
-    private String _replyQueue;
     /** Consumer attached to our reply queue */
     private DefaultConsumer _consumer;
 
@@ -73,7 +71,6 @@ public class RpcClient {
      * @param routingKey the routing key
      * @param timeout milliseconds before timing out on wait for response
      * @throws IOException if an error is encountered
-     * @see #setupReplyQueue
      */
     public RpcClient(Channel channel, String exchange, String routingKey, int timeout) throws IOException {
         _channel = channel;
@@ -83,7 +80,6 @@ public RpcClient(Channel channel, String exchange, String routingKey, int timeou
         _timeout = timeout;
         _correlationId = 0;
 
-        _replyQueue = setupReplyQueue();
         _consumer = setupConsumer();
     }
 
@@ -98,7 +94,6 @@ public RpcClient(Channel channel, String exchange, String routingKey, int timeou
      * @param exchange the exchange to connect to
      * @param routingKey the routing key
      * @throws IOException if an error is encountered
-     * @see #setupReplyQueue
      */
     public RpcClient(Channel channel, String exchange, String routingKey) throws IOException {
         this(channel, exchange, routingKey, NO_TIMEOUT);
@@ -125,16 +120,6 @@ public void close() throws IOException {
         }
     }
 
-    /**
-     * Creates a server-named exclusive autodelete queue to use for
-     * receiving replies to RPC requests.
-     * @throws IOException if an error is encountered
-     * @return the name of the reply queue
-     */
-    protected String setupReplyQueue() throws IOException {
-        return _channel.queueDeclare("", false, true, true, null).getQueue();
-    }
-
     /**
      * Registers a consumer on the reply queue.
      * @throws IOException if an error is encountered
@@ -167,7 +152,7 @@ public void handleDelivery(String consumerTag,
                 }
             }
         };
-        _channel.basicConsume(_replyQueue, true, consumer);
+        _channel.basicConsume("amq.rabbitmq.reply-to", true, consumer);
         return consumer;
     }
 
@@ -186,7 +171,7 @@ public byte[] primitiveCall(AMQP.BasicProperties props, byte[] message)
             _correlationId++;
             String replyId = "" + _correlationId;
             props = ((props==null) ? new AMQP.BasicProperties.Builder() : props.builder())
-                    .correlationId(replyId).replyTo(_replyQueue).build();
+                    .correlationId(replyId).replyTo("amq.rabbitmq.reply-to").build();
             _continuationMap.put(replyId, k);
         }
         publish(props, message);
@@ -332,14 +317,6 @@ public int getCorrelationId() {
         return _correlationId;
     }
 
-    /**
-     * Retrieve the reply queue.
-     * @return the name of the client's reply queue
-     */
-    public String getReplyQueue() {
-        return _replyQueue;
-    }
-
     /**
      * Retrieve the consumer.
      * @return an interface to the client's consumer object

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

@@ -17,7 +17,7 @@ public class ConnectionParams {
     private final int requestedChannelMax;
     private final int requestedHeartbeat;
     private final SaslConfig saslConfig;
-    private final int networkRecoveryInterval;
+    private final long networkRecoveryInterval;
     private final boolean topologyRecovery;
 
     private ExceptionHandler exceptionHandler;
@@ -41,7 +41,7 @@ public class ConnectionParams {
     public ConnectionParams(String username, String password, ExecutorService executor,
                             String virtualHost, Map<String, Object> clientProperties,
                             int requestedFrameMax, int requestedChannelMax, int requestedHeartbeat,
-                            SaslConfig saslConfig, int networkRecoveryInterval,
+                            SaslConfig saslConfig, long networkRecoveryInterval,
                             boolean topologyRecovery, ExceptionHandler exceptionHandler, ThreadFactory threadFactory) {
         this.username = username;
         this.password = password;
@@ -98,7 +98,7 @@ public ExceptionHandler getExceptionHandler() {
         return exceptionHandler;
     }
 
-    public int getNetworkRecoveryInterval() {
+    public long getNetworkRecoveryInterval() {
         return networkRecoveryInterval;
     }
 

src/com/rabbitmq/client/impl/recovery/AutorecoveringChannel.java

@@ -14,6 +14,7 @@
 import com.rabbitmq.client.ReturnListener;
 import com.rabbitmq.client.ShutdownListener;
 import com.rabbitmq.client.ShutdownSignalException;
+import com.rabbitmq.client.impl.AMQImpl;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -221,11 +222,11 @@ public AMQP.Exchange.DeleteOk exchangeDelete(String exchange) throws IOException
     }
 
     public AMQP.Exchange.BindOk exchangeBind(String destination, String source, String routingKey) throws IOException {
-        return delegate.exchangeBind(destination, source, routingKey);
+        return exchangeBind(destination, source, routingKey, null);
     }
 
     public AMQP.Exchange.BindOk exchangeBind(String destination, String source, String routingKey, Map<String, Object> arguments) throws IOException {
-        AMQP.Exchange.BindOk ok = delegate.exchangeBind(destination, source, routingKey, arguments);
+        final AMQP.Exchange.BindOk ok = delegate.exchangeBind(destination, source, routingKey, arguments);
         recordExchangeBinding(destination, source, routingKey, arguments);
         return ok;
     }
@@ -241,6 +242,7 @@ public AMQP.Exchange.UnbindOk exchangeUnbind(String destination, String source,
 
     public AMQP.Exchange.UnbindOk exchangeUnbind(String destination, String source, String routingKey, Map<String, Object> arguments) throws IOException {
         deleteRecordedExchangeBinding(destination, source, routingKey, arguments);
+        this.maybeDeleteRecordedAutoDeleteExchange(source);
         return delegate.exchangeUnbind(destination, source, routingKey, arguments);
     }
 
@@ -321,6 +323,7 @@ public AMQP.Queue.UnbindOk queueUnbind(String queue, String exchange, String rou
 
     public AMQP.Queue.UnbindOk queueUnbind(String queue, String exchange, String routingKey, Map<String, Object> arguments) throws IOException {
         deleteRecordedQueueBinding(queue, exchange, routingKey, arguments);
+        this.maybeDeleteRecordedAutoDeleteExchange(exchange);
         return delegate.queueUnbind(queue, exchange, routingKey, arguments);
     }
 
@@ -367,7 +370,8 @@ public String basicConsume(String queue, boolean autoAck, String consumerTag, bo
     }
 
     public void basicCancel(String consumerTag) throws IOException {
-        this.deleteRecordedConsumer(consumerTag);
+        RecordedConsumer c = this.deleteRecordedConsumer(consumerTag);
+        this.maybeDeleteRecordedAutoDeleteQueue(c.getQueue());
         delegate.basicCancel(consumerTag);
     }
 
@@ -578,7 +582,15 @@ private void recordConsumer(String result,
         this.connection.recordConsumer(result, consumer);
     }
 
-    private void deleteRecordedConsumer(String consumerTag) {
-        this.connection.deleteRecordedConsumer(consumerTag);
+    private RecordedConsumer deleteRecordedConsumer(String consumerTag) {
+        return this.connection.deleteRecordedConsumer(consumerTag);
+    }
+
+    private void maybeDeleteRecordedAutoDeleteQueue(String queue) {
+        this.connection.maybeDeleteRecordedAutoDeleteQueue(queue);
+    }
+
+    private void maybeDeleteRecordedAutoDeleteExchange(String exchange) {
+        this.connection.maybeDeleteRecordedAutoDeleteExchange(exchange);
     }
 }