QPID-578 : Queue Deletion causes unacked msgs to be discard resulting in the RejcectHandler throwing NPE when it too tries to discard the message.

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/M2.1@583415 13f79535-47bb-0310-9956-ffa450edef68

3 files changed, 28 insertions, 4 deletions

diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/BasicRejectMethodHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/BasicRejectMethodHandler.java
index 9052b2e81f..aca485f62a 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/handler/BasicRejectMethodHandler.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/handler/BasicRejectMethodHandler.java
@@ -23,7 +23,6 @@ package org.apache.qpid.server.handler;
 import org.apache.qpid.AMQException;
 import org.apache.qpid.framing.BasicRejectBody;
 import org.apache.qpid.protocol.AMQMethodEvent;
-import org.apache.qpid.protocol.AMQConstant;
 import org.apache.qpid.server.AMQChannel;
 import org.apache.qpid.server.ack.UnacknowledgedMessage;
 import org.apache.qpid.server.protocol.AMQProtocolSession;
@@ -71,7 +70,7 @@ public class BasicRejectMethodHandler implements StateAwareMethodListener<BasicR
         {
             _logger.debug("Rejecting:" + evt.getMethod().deliveryTag +
                           ": Requeue:" + evt.getMethod().requeue +
-//                              ": Resend:" + evt.getMethod().resend +
+                          //": Resend:" + evt.getMethod().resend +
                           " on channel:" + channel.debugIdentity());
         }
 
@@ -86,17 +85,31 @@ public class BasicRejectMethodHandler implements StateAwareMethodListener<BasicR
         }
         else
         {
+            if (message.queue == null || message.queue.isDeleted())
+            {
+                _logger.warn("Message's Queue as already been purged, unable to Reject. " +
+                             "Dropping message should use Dead Letter Queue");
+                //sendtoDeadLetterQueue(msg)                
+                return;
+            }
+
+            if (!message.message.isReferenced())
+            {
+                _logger.warn("Message as already been purged, unable to Reject.");
+                return;
+            }
+
 
             if (_logger.isTraceEnabled())
             {
                 _logger.trace("Rejecting: DT:" + deliveryTag + "-" + message.message.debugIdentity() +
                               ": Requeue:" + evt.getMethod().requeue +
-//                              ": Resend:" + evt.getMethod().resend +
+                              //": Resend:" + evt.getMethod().resend +
                               " on channel:" + channel.debugIdentity());
             }
 
             // If we haven't requested message to be resent to this consumer then reject it from ever getting it.
-//            if (!evt.getMethod().resend)
+            //if (!evt.getMethod().resend)
             {
                 message.message.reject(message.message.getDeliveredSubscription(message.queue));
             }
@@ -108,6 +121,7 @@ public class BasicRejectMethodHandler implements StateAwareMethodListener<BasicR
             else
             {
                 _logger.warn("Dropping message as requeue not required and there is no dead letter queue");
+                //sendtoDeadLetterQueue(AMQMessage message)
 //                message.queue = channel.getDefaultDeadLetterQueue();
 //                channel.requeue(deliveryTag);
             }
diff --git a/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java b/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java
index cd8c0198f3..01d0d870d7 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java
@@ -130,6 +130,11 @@ public class AMQMessage
 
     }
 
+    public boolean isReferenced()
+    {
+        return _referenceCount.get() > 0;
+    }    
+
     /**
      * Used to iterate through all the body frames associated with this message. Will not keep all the data in memory
      * therefore is memory-efficient.
diff --git a/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java b/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
index 2ea0b6d3d8..0c52a358f7 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
@@ -238,6 +238,11 @@ public class AMQQueue implements Managable, Comparable
         return _autoDelete;
     }
 
+    public boolean isDeleted()
+    {
+        return _deleted.get();
+    }    
+
     /** @return no of messages(undelivered) on the queue. */
     public int getMessageCount()
     {