diff options
author | Martin Ritchie <ritchiem@apache.org> | 2007-10-10 10:41:34 +0000 |
---|---|---|
committer | Martin Ritchie <ritchiem@apache.org> | 2007-10-10 10:41:34 +0000 |
commit | 7289ff94c6520175f1dbad689f245fda7f6c7220 (patch) | |
tree | ef020d0cdcee9e66f65f00453beb549f1b7e67b3 | |
parent | 3aa237a8e948c18c64ec3df25bd37bd719525eaf (diff) | |
download | qpid-python-7289ff94c6520175f1dbad689f245fda7f6c7220.tar.gz |
Merged revisions 573738-573739,573741-574077,574079-574236,574238-574265,574267-574503,574505-574554,574556-574584,574586-574873,574875-574901,574903-575737,575739-575787,575789-575810,575812-577772,577774-577940,577942-578057,578059-578732,578734,578736-578744,578746-578827,578829-578844,578846-579114,579116-579146,579148-579197,579199-579228,579230-579573,579575-579576,579579-579601,579603-579613,579615-579708,579710-580021,580023-580039,580042-580060,580062-580065,580067-580080,580082-580257,580259-580264,580266-580350,580352-580984,580986-580991,580994-581001,581003-581170,581172-581188,581190-581206,581208-581245,581247-581292,581294-581539,581541-581565,581567-581620,581622-581626,581628-581646,581648-581967,581969-582197,582199-582200,582203-582204,582206-582262,582264,582267-583084,583087,583089-583104,583106-583146,583148-583153,583155-583169,583171-583172,583174-583398,583400-583418 via svnmerge from
https://svn.apache.org/repos/asf/incubator/qpid/branches/M2.1
........
r583415 | ritchiem | 2007-10-10 10:59:56 +0100 (Wed, 10 Oct 2007) | 1 line
QPID-578 : Queue Deletion causes unacked msgs to be discard resulting in the RejcectHandler throwing NPE when it too tries to discard the message.
........
r583418 | ritchiem | 2007-10-10 11:11:13 +0100 (Wed, 10 Oct 2007) | 1 line
QPID-627 : Update provided by Aidan Skinner to correctly close resources used by Large Message Test.
........
git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/M2@583422 13f79535-47bb-0310-9956-ffa450edef68
4 files changed, 30 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() { diff --git a/java/client/src/test/java/org/apache/qpid/test/unit/basic/LargeMessageTest.java b/java/client/src/test/java/org/apache/qpid/test/unit/basic/LargeMessageTest.java index 55654fa522..cb030f3bef 100644 --- a/java/client/src/test/java/org/apache/qpid/test/unit/basic/LargeMessageTest.java +++ b/java/client/src/test/java/org/apache/qpid/test/unit/basic/LargeMessageTest.java @@ -68,6 +68,8 @@ public class LargeMessageTest extends TestCase protected void tearDown() throws Exception { super.tearDown(); + _session.close(); + _connection.close(); TransportConnection.killAllVMBrokers(); } |