Author: ritchiem
Date: Mon Oct 1 08:31:00 2007
New Revision: 580993
URL: http://svn.apache.org/viewvc?rev=580993&view=rev
Log:
QPID-611 QPID-620. DurableSubscriptionTest was failing due to a race condition
when using NO_ACK. This is due to the Queue Total Size being updated after the
send, but after the send and NO_ACK the msg data is purged and so there is no
size to retrieve. Changed all references to msg.dequeue to queue.dequeue where
appropriate so we can use that single point in the future for updating the
Queue Total Size.
Modified:
incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessage.java
incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java
incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/queue/ConcurrentSelectorDeliveryManager.java
incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionImpl.java
incubator/qpid/branches/M2.1/java/client/src/test/java/org/apache/qpid/test/unit/topic/DurableSubscriptionTest.java
Modified:
incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessage.java
URL:
http://svn.apache.org/viewvc/incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessage.java?rev=580993&r1=580992&r2=580993&view=diff
==============================================================================
---
incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessage.java
(original)
+++
incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessage.java
Mon Oct 1 08:31:00 2007
@@ -60,7 +60,7 @@
{
if (queue != null)
{
- message.dequeue(storeContext, queue);
+ queue.dequeue(storeContext, message);
}
//if the queue is null then the message is waiting to be acked, but
has been removed.
message.decrementReference(storeContext);
Modified:
incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java
URL:
http://svn.apache.org/viewvc/incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java?rev=580993&r1=580992&r2=580993&view=diff
==============================================================================
---
incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java
(original)
+++
incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java
Mon Oct 1 08:31:00 2007
@@ -592,7 +592,19 @@
_transientMessageData.addDestinationQueue(queue);
}
- public void dequeue(StoreContext storeContext, AMQQueue queue) throws
AMQException
+ /**
+ * NOTE: Think about why you are using this method. Normal usages would
want to do
+ * AMQQueue.dequeue(StoreContext, AMQMessage)
+ * This will keep the queue statistics up-to-date.
+ * Currently this method is only called _correctly_ from AMQQueue dequeue.
+ * Ideally we would have a better way for the queue to dequeue the message.
+ * Especially since enqueue isn't the recipriocal of this method.
+ * @deprecated
+ * @param storeContext
+ * @param queue
+ * @throws AMQException
+ */
+ void dequeue(StoreContext storeContext, AMQQueue queue) throws AMQException
{
_messageHandle.dequeue(storeContext, _messageId, queue);
}
Modified:
incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
URL:
http://svn.apache.org/viewvc/incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java?rev=580993&r1=580992&r2=580993&view=diff
==============================================================================
---
incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
(original)
+++
incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
Mon Oct 1 08:31:00 2007
@@ -809,7 +809,7 @@
}
}
- void dequeue(StoreContext storeContext, AMQMessage msg) throws
FailedDequeueException
+ public void dequeue(StoreContext storeContext, AMQMessage msg) throws
FailedDequeueException
{
try
{
Modified:
incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/queue/ConcurrentSelectorDeliveryManager.java
URL:
http://svn.apache.org/viewvc/incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/queue/ConcurrentSelectorDeliveryManager.java?rev=580993&r1=580992&r2=580993&view=diff
==============================================================================
---
incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/queue/ConcurrentSelectorDeliveryManager.java
(original)
+++
incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/queue/ConcurrentSelectorDeliveryManager.java
Mon Oct 1 08:31:00 2007
@@ -413,14 +413,16 @@
AMQMessage message = _messages.poll();
- message.dequeue(storeContext, queue);
-
- message.decrementReference(storeContext);
-
if (message != null)
{
+ queue.dequeue(storeContext, message);
+
_totalMessageSize.addAndGet(-message.getSize());
- }
+
+ //If this causes ref count to hit zero then data will be purged so
message.getSize() will NPE.
+ message.decrementReference(storeContext);
+
+ }
_lock.unlock();
}
@@ -485,7 +487,7 @@
_totalMessageSize.addAndGet(-message.getSize());
// Use the reapingStoreContext as any sub(if we have one) may
be in a tx.
- message.dequeue(_reapingStoreContext, _queue);
+ _queue.dequeue(_reapingStoreContext, message);
message.decrementReference(_reapingStoreContext);
@@ -511,13 +513,16 @@
}
/**
- * This method will return true if the message is to be purged from the
queue.
+ * This method will return true if the message is to be purged from the
queue.
*
*
- * SIDE-EFFECT: The message will be taken by the Subscription(sub) for
the current Queue(_queue)
+ * SIDE-EFFECT: The message will be taken by the Subscription(sub) for the
current Queue(_queue)
+ *
* @param message
* @param sub
+ *
* @return
+ *
* @throws AMQException
*/
private boolean purgeMessage(AMQMessage message, Subscription sub) throws
AMQException
@@ -607,6 +612,12 @@
") to :" + System.identityHashCode(sub));
}
+
+ if (messageQueue == _messages)
+ {
+ _totalMessageSize.addAndGet(-message.getSize());
+ }
+
sub.send(message, _queue);
//remove sent message from our queue.
@@ -654,10 +665,6 @@
}
}
- if ((message != null) && (messageQueue == _messages))
- {
- _totalMessageSize.addAndGet(-message.getSize());
- }
}
catch (AMQException e)
{
Modified:
incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionImpl.java
URL:
http://svn.apache.org/viewvc/incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionImpl.java?rev=580993&r1=580992&r2=580993&view=diff
==============================================================================
---
incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionImpl.java
(original)
+++
incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionImpl.java
Mon Oct 1 08:31:00 2007
@@ -307,12 +307,11 @@
}
protocolSession.getProtocolOutputConverter().writeDeliver(msg,
channel.getChannelId(), deliveryTag, consumerTag);
-
+
if (!_acks)
{
msg.decrementReference(storeContext);
}
-
}
}
finally
Modified:
incubator/qpid/branches/M2.1/java/client/src/test/java/org/apache/qpid/test/unit/topic/DurableSubscriptionTest.java
URL:
http://svn.apache.org/viewvc/incubator/qpid/branches/M2.1/java/client/src/test/java/org/apache/qpid/test/unit/topic/DurableSubscriptionTest.java?rev=580993&r1=580992&r2=580993&view=diff
==============================================================================
---
incubator/qpid/branches/M2.1/java/client/src/test/java/org/apache/qpid/test/unit/topic/DurableSubscriptionTest.java
(original)
+++
incubator/qpid/branches/M2.1/java/client/src/test/java/org/apache/qpid/test/unit/topic/DurableSubscriptionTest.java
Mon Oct 1 08:31:00 2007
@@ -114,18 +114,28 @@
con.close();
}
- public void testDurability() throws AMQException, JMSException,
URLSyntaxException
+ public void testDurabilityNOACK() throws AMQException, JMSException,
URLSyntaxException
+ {
+ durabilityImpl(AMQSession.NO_ACKNOWLEDGE);
+ }
+
+ public void testDurabilityAUTOACK() throws AMQException, JMSException,
URLSyntaxException
+ {
+ durabilityImpl(Session.AUTO_ACKNOWLEDGE);
+ }
+
+ private void durabilityImpl(int ackMode) throws AMQException,
JMSException, URLSyntaxException
{
AMQConnection con = new AMQConnection("vm://:1", "guest", "guest",
"test", "test");
AMQTopic topic = new AMQTopic(con, "MyTopic");
- Session session1 = con.createSession(false, AMQSession.NO_ACKNOWLEDGE);
+ Session session1 = con.createSession(false, ackMode);
MessageConsumer consumer1 = session1.createConsumer(topic);
- Session sessionProd = con.createSession(false,
AMQSession.NO_ACKNOWLEDGE);
+ Session sessionProd = con.createSession(false, ackMode);
MessageProducer producer = sessionProd.createProducer(topic);
- Session session2 = con.createSession(false, AMQSession.NO_ACKNOWLEDGE);
+ Session session2 = con.createSession(false, ackMode);
TopicSubscriber consumer2 = session2.createDurableSubscriber(topic,
"MySubscription");
con.start();
@@ -133,36 +143,41 @@
producer.send(session1.createTextMessage("A"));
Message msg;
- msg = consumer1.receive();
- assertEquals("A", ((TextMessage) msg).getText());
msg = consumer1.receive(100);
- assertEquals(null, msg);
+ assertNotNull("Message should be available", msg);
+ assertEquals("Message Text doesn't match", "A", ((TextMessage)
msg).getText());
+
+ msg = consumer1.receive(100);
+ assertNull("There should be no more messages for consumption on
consumer1.", msg);
msg = consumer2.receive();
- assertEquals("A", ((TextMessage) msg).getText());
+ assertNotNull(msg);
+ assertEquals("Consumer 2 should also received the first msg.", "A",
((TextMessage) msg).getText());
msg = consumer2.receive(100);
- assertEquals(null, msg);
+ assertNull("There should be no more messages for consumption on
consumer2.", msg);
consumer2.close();
- Session session3 = con.createSession(false, AMQSession.NO_ACKNOWLEDGE);
+ Session session3 = con.createSession(false, ackMode);
MessageConsumer consumer3 = session3.createDurableSubscriber(topic,
"MySubscription");
producer.send(session1.createTextMessage("B"));
_logger.info("Receive message on consumer 1 :expecting B");
msg = consumer1.receive(100);
- assertEquals("B", ((TextMessage) msg).getText());
+ assertNotNull("Consumer 1 should get message 'B'.", msg);
+ assertEquals("Incorrect Message recevied on consumer1.", "B",
((TextMessage) msg).getText());
_logger.info("Receive message on consumer 1 :expecting null");
msg = consumer1.receive(100);
- assertEquals(null, msg);
+ assertNull("There should be no more messages for consumption on
consumer1.", msg);
_logger.info("Receive message on consumer 3 :expecting B");
msg = consumer3.receive(100);
- assertEquals("B", ((TextMessage) msg).getText());
+ assertNotNull("Consumer 3 should get message 'B'.", msg);
+ assertEquals("Incorrect Message recevied on consumer4.", "B",
((TextMessage) msg).getText());
_logger.info("Receive message on consumer 3 :expecting null");
msg = consumer3.receive(100);
- assertEquals(null, msg);
+ assertNull("There should be no more messages for consumption on
consumer3.", msg);
con.close();
}
