Merge from default

Author: Simon MacMullen

src/com/rabbitmq/client/AckListener.java

@@ -0,0 +1,43 @@
+//   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-2010 LShift
+//   Ltd. Portions created by Cohesive Financial Technologies LLC are
+//   Copyright (C) 2007-2010 Cohesive Financial Technologies
+//   LLC. Portions created by Rabbit Technologies Ltd are Copyright
+//   (C) 2007-2010 Rabbit Technologies Ltd.
+//
+//   All Rights Reserved.
+//
+//   Contributor(s): ______________________________________.
+//
+
+package com.rabbitmq.client;
+
+import java.io.IOException;
+
+/**
+ * Implement this interface in order to be notified of Basic.Ack
+ * events.
+ */
+public interface AckListener {
+    void handleAck(long deliveryTag, boolean multiple)
+        throws IOException;
+}

src/com/rabbitmq/client/Channel.java

@@ -38,6 +38,7 @@
 import com.rabbitmq.client.AMQP.Queue;
 import com.rabbitmq.client.AMQP.Tx;
 import com.rabbitmq.client.AMQP.Basic;
+import com.rabbitmq.client.AMQP.Confirm;
 import com.rabbitmq.client.AMQP.Channel.FlowOk;
 
 /**
@@ -147,6 +148,18 @@ public interface Channel extends ShutdownNotifier {
      */
     void setFlowListener(FlowListener listener);
 
+    /**
+     * Return the current {@link AckListener}.
+     * @return an interface to the current ack listener.
+     */
+    AckListener getAckListener();
+
+    /**
+     * Set the current {@link AckListener}.
+     * @param listener the listener to use, or null indicating "don't use one".
+     */
+    void setAckListener(AckListener listener);
+
     /**
      * Get the current default consumer. @see setDefaultConsumer for rationale.
      * @return an interface to the current default consumer.
@@ -585,6 +598,17 @@ Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, b
      */
     void basicCancel(String consumerTag) throws IOException;
 
+    /**
+     * Ask the broker to resend unacknowledged messages.  In 0-8
+     * basic.recover is asynchronous; in 0-9-1 it is synchronous, and
+     * the new, deprecated method basic.recover_async is asynchronous.
+     * <p/>
+     * Equivalent to calling <code>basicRecover(true)</code>, messages 
+     * will be requeued and possibly delivered to a different consumer. 
+     * @see #basicRecover(boolean)
+     */
+     Basic.RecoverOk basicRecover() throws IOException;
+  
     /**
      * Ask the broker to resend unacknowledged messages.  In 0-8
      * basic.recover is asynchronous; in 0-9-1 it is synchronous, and
@@ -633,4 +657,21 @@ Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, b
      * @throws java.io.IOException if an error is encountered
      */
     Tx.RollbackOk txRollback() throws IOException;
+
+    /**
+     * Enables publisher acknowledgements on this channel.
+     * @param multiple determines whether the broker can acknowledge
+     * multiple messages at the same time
+     * @see com.rabbitmq.client.AMQP.Confirm.Select
+     * @throws java.io.IOException if an error is encountered
+     */
+    Confirm.SelectOk confirmSelect(boolean multiple) throws IOException;
+
+    /**
+     * Returns the number of messages published since the channel was
+     * put in confirm mode.
+     * @return the number of messages published since the first
+     * confirm.select; if the channel is not in confirm mode, -1 is
+     * returned */
+    long getPublishedMessageCount();
 }

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

@@ -227,8 +227,8 @@ public AMQConnection(ConnectionFactory factory,
      * calls Connection.Open and waits for the OpenOk. Sets heartbeat
      * and frame max values after tuning has taken place.
      * @throws java.io.IOException if an error is encountered; IOException
-     * subtypes ProtocolVersionMismatchException and
-     * PossibleAuthenticationFailureException will be thrown in the
+     * subtypes {@link ProtocolVersionMismatchException} and
+     * {@link PossibleAuthenticationFailureException} will be thrown in the
      * corresponding circumstances.
      */
     public void start()

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

@@ -31,6 +31,7 @@
 
 package com.rabbitmq.client.impl;
 
+import com.rabbitmq.client.AckListener;
 import com.rabbitmq.client.AMQP.BasicProperties;
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.Command;
@@ -45,6 +46,7 @@
 import com.rabbitmq.client.UnexpectedMethodError;
 import com.rabbitmq.client.impl.AMQImpl.Basic;
 import com.rabbitmq.client.impl.AMQImpl.Channel;
+import com.rabbitmq.client.impl.AMQImpl.Confirm;
 import com.rabbitmq.client.impl.AMQImpl.Exchange;
 import com.rabbitmq.client.impl.AMQImpl.Queue;
 import com.rabbitmq.client.impl.AMQImpl.Tx;
@@ -54,6 +56,7 @@
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.TimeoutException;
 
 
@@ -98,6 +101,14 @@ public class ChannelN extends AMQChannel implements com.rabbitmq.client.Channel
      */
     public volatile FlowListener flowListener = null;
 
+    /** Reference to the currently-active AckListener, or null if there is none.
+     */
+    public volatile AckListener ackListener = null;
+
+    /** Current published message count (used by publisher acknowledgements)
+     */
+    private final AtomicLong publishedMessageCount = new AtomicLong(-1);
+
     /** Reference to the currently-active default consumer, or null if there is
      *  none.
      */
@@ -153,6 +164,19 @@ public void setFlowListener(FlowListener listener) {
         flowListener = listener;
     }
 
+    /** Returns the current AckListener. */
+    public AckListener getAckListener() {
+        return ackListener;
+    }
+
+    /**
+     * Sets the current AckListener.
+     * A null argument is interpreted to mean "do not use an ack listener".
+     */
+    public void setAckListener(AckListener listener) {
+        ackListener = listener;
+    }
+
     /** Returns the current default consumer. */
     public Consumer getDefaultConsumer() {
         return defaultConsumer;
@@ -310,6 +334,17 @@ public void releaseChannelNumber() {
                     }
                 }
                 return true;
+            } else if (method instanceof Basic.Ack) {
+                Basic.Ack ack = (Basic.Ack) method;
+                AckListener l = getAckListener();
+                if (l != null) {
+                    try {
+                        l.handleAck(ack.getDeliveryTag(), ack.getMultiple());
+                    } catch (Throwable ex) {
+                        _connection.getExceptionHandler().handleAckListenerException(this, ex);
+                    }
+                }
+                return true;
             } else if (method instanceof Basic.RecoverOk) {
                 for (Consumer callback: _consumers.values()) {
                     callback.handleRecoverOk();
@@ -463,6 +498,8 @@ public void basicPublish(String exchange, String routingKey,
                              BasicProperties props, byte[] body)
         throws IOException
     {
+        if (publishedMessageCount.get() >= 0)
+            publishedMessageCount.incrementAndGet();
         BasicProperties useProps = props;
         if (props == null) {
             useProps = MessageProperties.MINIMAL_BASIC;
@@ -553,7 +590,7 @@ public Exchange.UnbindOk exchangeUnbind(String destination, String source,
             String routingKey) throws IOException {
         return exchangeUnbind(destination, source, routingKey, null);
     }
-    
+
     /** Public API - {@inheritDoc} */
     public Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive,
                                         boolean autoDelete, Map<String, Object> arguments)
@@ -769,6 +806,14 @@ public Consumer transformReply(AMQCommand replyCommand) {
         }
     }
 
+
+     /** Public API - {@inheritDoc} */
+    public Basic.RecoverOk basicRecover()
+        throws IOException
+    {
+        return basicRecover(true);
+    }
+
      /** Public API - {@inheritDoc} */
     public Basic.RecoverOk basicRecover(boolean requeue)
         throws IOException
@@ -805,6 +850,17 @@ public Tx.RollbackOk txRollback()
         return (Tx.RollbackOk) exnWrappingRpc(new Tx.Rollback()).getMethod();
     }
 
+    /** Public API - {@inheritDoc} */
+    public Confirm.SelectOk confirmSelect(boolean multiple)
+        throws IOException
+    {
+        if (publishedMessageCount.get() == -1)
+            publishedMessageCount.set(0);
+        return (Confirm.SelectOk)
+            exnWrappingRpc(new Confirm.Select(multiple, false)).getMethod();
+
+    }
+
     /** Public API - {@inheritDoc} */
     public Channel.FlowOk flow(final boolean a) throws IOException {
         return (Channel.FlowOk) exnWrappingRpc(new Channel.Flow() {{active = a;}}).getMethod();
@@ -815,4 +871,8 @@ public Channel.FlowOk getFlow() {
         return new Channel.FlowOk(!_blockContent);
     }
 
+    /** Public API - {@inheritDoc} */
+    public long getPublishedMessageCount() {
+        return publishedMessageCount.longValue();
+    }
 }

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

@@ -55,6 +55,10 @@ public void handleFlowListenerException(Channel channel, Throwable exception) {
         handleChannelKiller(channel, exception, "FlowListener.handleFlow");
     }
 
+    public void handleAckListenerException(Channel channel, Throwable exception) {
+        handleChannelKiller(channel, exception, "AckListener.handleAck");
+    }
+
     public void handleConsumerException(Channel channel, Throwable exception,
                                         Consumer consumer, String consumerTag,
                                         String methodName)

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

@@ -59,14 +59,24 @@ public interface ExceptionHandler {
 
     /**
      * Perform any required exception processing for the situation
-     * when the driver thread for the connection has called a 
+     * when the driver thread for the connection has called a
      * FlowListener's handleFlow method, and that method has
      * thrown an exeption.
      * @param channel the ChannelN that held the FlowListener
      * @param exception the exception thrown by FlowListener.handleFlow
      */
     void handleFlowListenerException(Channel channel, Throwable exception);
 
+    /**
+     * Perform any required exception processing for the situation
+     * when the driver thread for the connection has called an
+     * AckListener's handleAck method, and that method has
+     * thrown an exeption.
+     * @param channel the ChannelN that held the AckListener
+     * @param exception the exception thrown by AckListener.handleAck
+     */
+    void handleAckListenerException(Channel channel, Throwable exception);
+
     /**
      * Perform any required exception processing for the situation
      * when the driver thread for the connection has called a method

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

@@ -201,6 +201,10 @@ public void handleFlowListenerException(Channel ch, Throwable ex) {
             fail("handleFlowListenerException: " + ex);
         }
 
+        public void handleAckListenerException(Channel ch, Throwable ex) {
+            fail("handleAckListenerException: " + ex);
+        }
+
         public void handleConsumerException(Channel ch,
                                             Throwable ex,
                                             Consumer c,