default to stable

Author: Simon MacMullen

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/Channel.java

@@ -343,6 +343,28 @@ Exchange.DeclareOk exchangeDeclare(String exchange,
                                               boolean internal,
                                               Map<String, Object> arguments) throws IOException;
 
+    /**
+     * Like {@link Channel#exchangeDeclare(String, String, boolean, boolean, java.util.Map)} but
+     * sets nowait parameter to true and returns nothing (as there will be no response from
+     * the server).
+     *
+     * @param exchange the name of the exchange
+     * @param type the exchange type
+     * @param durable true if we are declaring a durable exchange (the exchange will survive a server restart)
+     * @param autoDelete true if the server should delete the exchange when it is no longer in use
+     * @param internal true if the exchange is internal, i.e. can't be directly
+     * published to by a client.
+     * @param arguments other properties (construction arguments) for the exchange
+     * @return a declaration-confirm method to indicate the exchange was successfully declared
+     * @throws java.io.IOException if an error is encountered
+     */
+    void exchangeDeclareNoWait(String exchange,
+                               String type,
+                               boolean durable,
+                               boolean autoDelete,
+                               boolean internal,
+                               Map<String, Object> arguments) throws IOException;
+
     /**
      * Declare an exchange passively; that is, check if the named exchange exists.
      * @param name check the existence of an exchange named this
@@ -361,6 +383,18 @@ Exchange.DeclareOk exchangeDeclare(String exchange,
      */
     Exchange.DeleteOk exchangeDelete(String exchange, boolean ifUnused) throws IOException;
 
+    /**
+     * Like {@link Channel#exchangeDelete(String, boolean)} but sets nowait parameter to true
+     * and returns void (as there will be no response from the server).
+     * @see com.rabbitmq.client.AMQP.Exchange.Delete
+     * @see com.rabbitmq.client.AMQP.Exchange.DeleteOk
+     * @param exchange the name of the exchange
+     * @param ifUnused true to indicate that the exchange is only to be deleted if it is unused
+     * @throws java.io.IOException if an error is encountered
+     */
+    void exchangeDeleteNoWait(String exchange, boolean ifUnused) throws IOException;
+
+
     /**
      * Delete an exchange, without regard for whether it is in use or not
      * @see com.rabbitmq.client.AMQP.Exchange.Delete
@@ -396,6 +430,17 @@ Exchange.DeclareOk exchangeDeclare(String exchange,
      */
     Exchange.BindOk exchangeBind(String destination, String source, String routingKey, Map<String, Object> arguments) throws IOException;
 
+    /**
+     * Like {@link Channel#exchangeBind(String, String, String, java.util.Map)} but sets nowait parameter
+     * to true and returns void (as there will be no response from the server).
+     * @param destination the name of the exchange to which messages flow across the binding
+     * @param source the name of the exchange from which messages flow across the binding
+     * @param routingKey the routine key to use for the binding
+     * @param arguments other properties (binding parameters)
+     * @throws java.io.IOException if an error is encountered
+     */
+    void exchangeBindNoWait(String destination, String source, String routingKey, Map<String, Object> arguments) throws IOException;
+
     /**
      * Unbind an exchange from an exchange, with no extra arguments.
      * @see com.rabbitmq.client.AMQP.Exchange.Bind
@@ -421,6 +466,17 @@ Exchange.DeclareOk exchangeDeclare(String exchange,
      */
     Exchange.UnbindOk exchangeUnbind(String destination, String source, String routingKey, Map<String, Object> arguments) throws IOException;
 
+    /**
+     * Same as {@link Channel#exchangeUnbind(String, String, String, java.util.Map)} but sets no-wait parameter to true
+     * and returns nothing (as there will be no response from the server).
+     * @param destination the name of the exchange to which messages flow across the binding
+     * @param source the name of the exchange from which messages flow across the binding
+     * @param routingKey the routine key to use for the binding
+     * @param arguments other properties (binding parameters)
+     * @throws java.io.IOException if an error is encountered
+     */
+    void exchangeUnbindNoWait(String destination, String source, String routingKey, Map<String, Object> arguments) throws IOException;
+
     /**
      * Actively declare a server-named exclusive, autodelete, non-durable queue.
      * The name of the new queue is held in the "queue" field of the {@link com.rabbitmq.client.AMQP.Queue.DeclareOk} result.
@@ -446,6 +502,19 @@ Exchange.DeclareOk exchangeDeclare(String exchange,
     Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,
                                  Map<String, Object> arguments) throws IOException;
 
+    /**
+     * Like {@link Channel#queueDeclare(String, boolean, boolean, boolean, java.util.Map)} but sets nowait
+     * flag to true and returns no result (as there will be no response from the server).
+     * @param queue the name of the queue
+     * @param durable true if we are declaring a durable queue (the queue will survive a server restart)
+     * @param exclusive true if we are declaring an exclusive queue (restricted to this connection)
+     * @param autoDelete true if we are declaring an autodelete queue (server will delete it when no longer in use)
+     * @param arguments other properties (construction arguments) for the queue
+     * @throws java.io.IOException if an error is encountered
+     */
+    void queueDeclareNoWait(String queue, boolean durable, boolean exclusive, boolean autoDelete,
+                            Map<String, Object> arguments) throws IOException;
+
     /**
      * Declare a queue passively; i.e., check if it exists.  In AMQP
      * 0-9-1, all arguments aside from nowait are ignored; and sending
@@ -482,6 +551,18 @@ Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, b
      */
     Queue.DeleteOk queueDelete(String queue, boolean ifUnused, boolean ifEmpty) throws IOException;
 
+    /**
+     * Like {@link Channel#queueDelete(String, boolean, boolean)} but sets nowait parameter
+     * to true and returns nothing (as there will be no response from the server).
+     * @see com.rabbitmq.client.AMQP.Queue.Delete
+     * @see com.rabbitmq.client.AMQP.Queue.DeleteOk
+     * @param queue the name of the queue
+     * @param ifUnused true if the queue should be deleted only if not in use
+     * @param ifEmpty true if the queue should be deleted only if empty
+     * @throws java.io.IOException if an error is encountered
+     */
+    void queueDeleteNoWait(String queue, boolean ifUnused, boolean ifEmpty) throws IOException;
+
     /**
      * Bind a queue to an exchange, with no extra arguments.
      * @see com.rabbitmq.client.AMQP.Queue.Bind
@@ -507,6 +588,18 @@ Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, b
      */
     Queue.BindOk queueBind(String queue, String exchange, String routingKey, Map<String, Object> arguments) throws IOException;
 
+    /**
+     * Same as {@link Channel#queueDeclare(String, boolean, boolean, boolean, java.util.Map)} but sets nowait
+     * parameter to true and returns void (as there will be no response
+     * from the server).
+     * @param queue the name of the queue
+     * @param exchange the name of the exchange
+     * @param routingKey the routine key to use for the binding
+     * @param arguments other properties (binding parameters)
+     * @throws java.io.IOException if an error is encountered
+     */
+    void queueBindNoWait(String queue, String exchange, String routingKey, Map<String, Object> arguments) throws IOException;
+
     /**
      * Unbinds a queue from an exchange, with no extra arguments.
      * @see com.rabbitmq.client.AMQP.Queue.Unbind

src/com/rabbitmq/client/ConnectionFactory.java

@@ -75,9 +75,12 @@ public class ConnectionFactory implements Cloneable {
     /** The default connection timeout;
      *  zero means wait indefinitely */
     public static final int    DEFAULT_CONNECTION_TIMEOUT = 0;
+    /** The default shutdown timeout;
+     *  zero means wait indefinitely */
+    public static final int    DEFAULT_SHUTDOWN_TIMEOUT = 10000;
 
     /** The default SSL protocol */
-    private static final String DEFAULT_SSL_PROTOCOL = "SSLv3";
+    private static final String DEFAULT_SSL_PROTOCOL = "TLSv1";
 
     private String username                       = DEFAULT_USER;
     private String password                       = DEFAULT_PASS;
@@ -88,6 +91,7 @@ public class ConnectionFactory implements Cloneable {
     private int requestedFrameMax                 = DEFAULT_FRAME_MAX;
     private int requestedHeartbeat                = DEFAULT_HEARTBEAT;
     private int connectionTimeout                 = DEFAULT_CONNECTION_TIMEOUT;
+    private int shutdownTimeout                   = DEFAULT_SHUTDOWN_TIMEOUT;
     private Map<String, Object> _clientProperties = AMQConnection.defaultClientProperties();
     private SocketFactory factory                 = SocketFactory.getDefault();
     private SaslConfig saslConfig                 = DefaultSaslConfig.PLAIN;
@@ -99,7 +103,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
@@ -330,6 +337,26 @@ public int getConnectionTimeout() {
         return this.connectionTimeout;
     }
 
+    /**
+     * Set the shutdown timeout. This is the amount of time that Consumer implementations have to
+     * continue working through deliveries (and other Consumer callbacks) <b>after</b> the connection
+     * has closed but before the ConsumerWorkService is torn down. If consumers exceed this timeout
+     * then any remaining queued deliveries (and other Consumer callbacks, <b>including</b>
+     * the Consumer's handleShutdownSignal() invocation) will be lost.
+     * @param shutdownTimeout shutdown timeout in milliseconds; zero for infinite; default 10000
+     */
+    public void setShutdownTimeout(int shutdownTimeout) {
+        this.shutdownTimeout = shutdownTimeout;
+    }
+
+    /**
+     * Retrieve the shutdown timeout.
+     * @return the shutdown timeout, in milliseconds; zero for infinite
+     */
+    public int getShutdownTimeout() {
+        return shutdownTimeout;
+    }
+
     /**
      * Set the requested heartbeat.
      * @param requestedHeartbeat the initially requested heartbeat interval, in seconds; zero for none
@@ -493,7 +520,7 @@ public void useSslProtocol(String protocol)
 
     /**
      * Convenience method for setting up an SSL socket factory.
-     * Pass in the SSL protocol to use, e.g. "TLS" or "SSLv3".
+     * Pass in the SSL protocol to use, e.g. "TLSv1" or "TLSv1.2".
      *
      * @param protocol SSL protocol to use.
      */
@@ -599,7 +626,7 @@ public Connection newConnection(ExecutorService executor, Address[] addrs)
 
     public ConnectionParams params(ExecutorService executor) {
         return new ConnectionParams(username, password, executor, virtualHost, getClientProperties(),
-                                    requestedFrameMax, requestedChannelMax, requestedHeartbeat, saslConfig,
+                                    requestedFrameMax, requestedChannelMax, requestedHeartbeat, shutdownTimeout, saslConfig,
                                     networkRecoveryInterval, topologyRecovery, exceptionHandler, threadFactory);
     }
 
@@ -638,7 +665,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 +676,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/AMQConnection.java

@@ -139,6 +139,7 @@ public static final Map<String, Object> defaultClientProperties() {
     private final int requestedHeartbeat;
     private final int requestedChannelMax;
     private final int requestedFrameMax;
+    private final int shutdownTimeout;
     private final String username;
     private final String password;
     private final Collection<BlockedListener> blockedListeners = new CopyOnWriteArrayList<BlockedListener>();
@@ -217,6 +218,7 @@ public AMQConnection(ConnectionParams params, FrameHandler frameHandler)
         this.requestedFrameMax = params.getRequestedFrameMax();
         this.requestedChannelMax = params.getRequestedChannelMax();
         this.requestedHeartbeat = params.getRequestedHeartbeat();
+        this.shutdownTimeout = params.getShutdownTimeout();
         this.saslConfig = params.getSaslConfig();
         this.executor = params.getExecutor();
         this.threadFactory = params.getThreadFactory();
@@ -229,7 +231,7 @@ public AMQConnection(ConnectionParams params, FrameHandler frameHandler)
     }
 
     private void initializeConsumerWorkService() {
-        this._workService  = new ConsumerWorkService(executor, threadFactory);
+        this._workService  = new ConsumerWorkService(executor, threadFactory, shutdownTimeout);
     }
 
     private void initializeHeartbeatSender() {