added ack listener to Java API

Author: Alexandru Scvortov

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 sequenceNumber)
+        throws IOException;
+}

src/com/rabbitmq/client/Channel.java

@@ -148,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.

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;
@@ -99,6 +100,10 @@ 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;
+
     /** Reference to the currently-active default consumer, or null if there is
      *  none.
      */
@@ -154,6 +159,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;
@@ -311,6 +329,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());
+                    } catch (Throwable ex) {
+                        _connection.getExceptionHandler().handleAckListenerException(this, ex);
+                    }
+                }
+                return true;
             } else if (method instanceof Basic.RecoverOk) {
                 for (Consumer callback: _consumers.values()) {
                     callback.handleRecoverOk();

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