add NackListener

Author: Alexandru Scvortov

src/com/rabbitmq/client/Channel.java

@@ -146,6 +146,18 @@ public interface Channel extends ShutdownNotifier {
      */
     void setAckListener(AckListener listener);
 
+    /**
+     * Return the current {@link NackListener}.
+     * @return an interface to the current nack listener.
+     */
+    NackListener getNackListener();
+
+    /**
+     * Set the current {@link NackListener}.
+     * @param listener the listener to use, or null indicating "don't use one".
+     */
+    void setNackListener(NackListener listener);
+
     /**
      * Get the current default consumer. @see setDefaultConsumer for rationale.
      * @return an interface to the current default consumer.

src/com/rabbitmq/client/NackListener.java

@@ -0,0 +1,29 @@
+//  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 Developer of the Original Code is VMware, Inc.
+//  Copyright (c) 2007-2011 VMware, Inc.  All rights reserved.
+//
+
+
+package com.rabbitmq.client;
+
+import java.io.IOException;
+
+/**
+ * Implement this interface in order to be notified of Basic.Nack
+ * events.
+ */
+public interface NackListener {
+    void handleNack(long deliveryTag, boolean multiple)
+        throws IOException;
+}

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

@@ -18,6 +18,7 @@
 package com.rabbitmq.client.impl;
 
 import com.rabbitmq.client.AckListener;
+import com.rabbitmq.client.NackListener;
 import com.rabbitmq.client.AMQP.BasicProperties;
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.Command;
@@ -91,6 +92,10 @@ public class ChannelN extends AMQChannel implements com.rabbitmq.client.Channel
      */
     public volatile AckListener ackListener = null;
 
+    /** Reference to the currently-active NackListener, or null if there is none.
+     */
+    public volatile NackListener nackListener = null;
+
     /** Sequence number of next published message requiring confirmation.
      */
     private long nextPublishSeqNo = 0L;
@@ -163,6 +168,19 @@ public void setAckListener(AckListener listener) {
         ackListener = listener;
     }
 
+    /** Returns the current NackListener. */
+    public NackListener getNackListener() {
+        return nackListener;
+    }
+
+    /**
+     * Sets the current NackListener.
+     * A null argument is interpreted to mean "do not use a nack listener".
+     */
+    public void setNackListener(NackListener listener) {
+        nackListener = listener;
+    }
+
     /** Returns the current default consumer. */
     public Consumer getDefaultConsumer() {
         return defaultConsumer;
@@ -331,6 +349,17 @@ public void releaseChannelNumber() {
                     }
                 }
                 return true;
+            } else if (method instanceof Basic.Nack) {
+                Basic.Nack nack = (Basic.Nack) method;
+                NackListener l = getNackListener();
+                if (l != null) {
+                    try {
+                        l.handleNack(nack.getDeliveryTag(), nack.getMultiple());
+                    } catch (Throwable ex) {
+                        _connection.getExceptionHandler().handleNackListenerException(this, ex);
+                    }
+                }
+                return true;
             } else if (method instanceof Basic.RecoverOk) {
                 for (Consumer callback: _consumers.values()) {
                     callback.handleRecoverOk();

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

@@ -45,6 +45,10 @@ public void handleAckListenerException(Channel channel, Throwable exception) {
         handleChannelKiller(channel, exception, "AckListener.handleAck");
     }
 
+    public void handleNackListenerException(Channel channel, Throwable exception) {
+        handleChannelKiller(channel, exception, "NackListener.handleNack");
+    }
+
     public void handleConsumerException(Channel channel, Throwable exception,
                                         Consumer consumer, String consumerTag,
                                         String methodName)

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

@@ -63,6 +63,16 @@ public interface ExceptionHandler {
      */
     void handleAckListenerException(Channel channel, Throwable exception);
 
+    /**
+     * Perform any required exception processing for the situation
+     * when the driver thread for the connection has called an
+     * NackListener's handleNack method, and that method has
+     * thrown an exeption.
+     * @param channel the ChannelN that held the NackListener
+     * @param exception the exception thrown by NackListener.handleNack
+     */
+    void handleNackListenerException(Channel channel, Throwable exception);
+
     /**
      * Perform any required exception processing for the situation
      * when the driver thread for the connection has called a method