add waitForConfirmsOrDie

Author: Alexandru Scvortov

src/com/rabbitmq/client/Channel.java

@@ -703,7 +703,14 @@ void basicNack(long deliveryTag, boolean multiple, boolean requeue)
      * non-Confirm channel, waitForConfirms returns true immediately.
      * @return whether all the messages were ack'd (and none were nack'd)
      */
-    boolean waitForConfirms() throws InterruptedException;
+    boolean waitForConfirms() throws IOException, InterruptedException;
+
+    /** Wait untill all messages published since the last call have
+     * been either ack'd or nack'd by the broker.  If any of the
+     * messages were nack'd, waitForConfirmsOrDie will throw an
+     * IOException.  When called on a non-Confirm channel, it will
+     * return immediately. */
+    void waitForConfirmsOrDie() throws IOException, InterruptedException;
 
     /**
      * Asynchronously send a method over this channel.

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

@@ -188,13 +188,31 @@ public ConfirmListener getConfirmListener() {
 
     /** {@inheritDoc} */
     public boolean waitForConfirms()
-        throws InterruptedException
+        throws IOException, InterruptedException
+    {
+        return waitForConfirmsInternal(false);
+    }
+
+    /** {@inheritDoc} */
+    public void waitForConfirmsOrDie()
+        throws IOException, InterruptedException
+    {
+        waitForConfirmsInternal(true);
+    }
+
+    protected boolean waitForConfirmsInternal(boolean dieOnNack)
+        throws IOException, InterruptedException
     {
         synchronized (unconfirmedSet) {
             while (true) {
                 if (getCloseReason() != null) {
                     throw Utility.fixStackTrace(getCloseReason());
                 }
+                if (dieOnNack && nacksReceived) {
+                    close(AMQP.REPLY_SUCCESS, "OK", true,
+                          new RuntimeException("received nack"), false);
+                    throw Utility.fixStackTrace(getCloseReason());
+                }
                 if (unconfirmedSet.isEmpty()) {
                     boolean noNacksReceived = !nacksReceived;
                     nacksReceived = false;