Author: robbie
Date: Wed Dec 15 16:07:08 2010
New Revision: 1049624
URL: http://svn.apache.org/viewvc?rev=1049624&view=rev
Log:
QPID-2980: expose the delivery count as a property
Modified:
qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_10.java
qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_8.java
qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/CustomJMSXProperty.java
qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/DeliveryCountTracker.java
qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate.java
qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_10.java
qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_8.java
qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessage.java
qpid/branches/0.5.x-dev/qpid/java/client/src/test/java/org/apache/qpid/client/DeliveryCountTrackerTest.java
qpid/branches/0.5.x-dev/qpid/java/client/src/test/java/org/apache/qpid/client/message/AbstractJMSMessageTest.java
qpid/branches/0.5.x-dev/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/MaxDeliveryCountTest.java
Modified:
qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
URL:
http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java?rev=1049624&r1=1049623&r2=1049624&view=diff
==============================================================================
---
qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
(original)
+++
qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
Wed Dec 15 16:07:08 2010
@@ -756,12 +756,16 @@ public abstract class BasicMessageConsum
msgId = message.getJMSMessageID();
if(msgId != null)
{
- _tracker.recordMessage(msgId, message.getDeliveryTag());
+ int count = _tracker.recordMessage(msgId,
message.getDeliveryTag());
+ if(count != 0)
+ {
+ message.setJMSXDeliveryCount(count);
+ }
}
}
catch (JMSException e)
{
- _logger.warn("Exception while retrieving JMSMessageID from
message" +
+ _logger.error("Exception while retrieving JMSMessageID from
message" +
" with deliveryTag '" + message.getDeliveryTag() + "': " +
e, e);
}
}
Modified:
qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_10.java
URL:
http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_10.java?rev=1049624&r1=1049623&r2=1049624&view=diff
==============================================================================
---
qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_10.java
(original)
+++
qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_10.java
Wed Dec 15 16:07:08 2010
@@ -82,6 +82,9 @@ public class BasicMessageProducer_0_10 e
// On the receiving side, this will be read in to the JMSXUserID as
well.
messageProps.setUserId(userIDBytes);
+
+ //ensure JMSXDeliverCount is cleared
+ delegate.setJMSXDeliveryCount(null);
if (messageId != null)
{
Modified:
qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_8.java
URL:
http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_8.java?rev=1049624&r1=1049623&r2=1049624&view=diff
==============================================================================
---
qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_8.java
(original)
+++
qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_8.java
Wed Dec 15 16:07:08 2010
@@ -125,6 +125,9 @@ public class BasicMessageProducer_0_8 ex
contentHeaderProperties.setDeliveryMode((byte) deliveryMode);
contentHeaderProperties.setPriority((byte) priority);
+
+ //ensure JMSXDeliveryCount is cleared
+ delegate.setJMSXDeliveryCount(null);
final int size = (payload != null) ? payload.limit() : 0;
final int contentBodyFrameCount =
calculateContentBodyFrameCount(payload);
Modified:
qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/CustomJMSXProperty.java
URL:
http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/CustomJMSXProperty.java?rev=1049624&r1=1049623&r2=1049624&view=diff
==============================================================================
---
qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/CustomJMSXProperty.java
(original)
+++
qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/CustomJMSXProperty.java
Wed Dec 15 16:07:08 2010
@@ -32,6 +32,7 @@ public enum CustomJMSXProperty
JMS_QPID_DESTTYPE,
JMSXGroupID,
JMSXGroupSeq,
+ JMSXDeliveryCount,
JMSXUserID;
Modified:
qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/DeliveryCountTracker.java
URL:
http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/DeliveryCountTracker.java?rev=1049624&r1=1049623&r2=1049624&view=diff
==============================================================================
---
qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/DeliveryCountTracker.java
(original)
+++
qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/DeliveryCountTracker.java
Wed Dec 15 16:07:08 2010
@@ -90,44 +90,46 @@ public class DeliveryCountTracker
*
* @param msgID the JMSMessageID of the message to track
* @param deliveryTag the delivery tag of the most recent encounter of the
message
+ * @return the count of how many times the message has now been seen, or 0
if unknown.
*/
- public synchronized void recordMessage(String msgID, long deliveryTag)
+ public synchronized int recordMessage(String msgID, long deliveryTag)
{
- try
+ int count = 0;
+
+ if(msgID == null)
{
- if(msgID == null)
- {
- //we can't distinguish between different
- //messages without a JMSMessageID, so skip
- return;
- }
+ //we can't distinguish between different
+ //messages without a JMSMessageID, so skip
+ return count;
+ }
- _jmsIDtoDeliverTag.put(msgID, deliveryTag);
+ _jmsIDtoDeliverTag.put(msgID, deliveryTag);
- Integer count = _receivedMsgIDs.get(msgID);
+ //using Integer to allow null check for the count map
+ Integer mapCount = _receivedMsgIDs.get(msgID);
- if(count != null)
- {
- ++count;
- if (_logger.isDebugEnabled())
- {
- _logger.debug("Incrementing count for JMSMessageID: '" +
msgID + "', value now: " + count);
- }
- _receivedMsgIDs.put(msgID, count);
- }
- else
+ if(mapCount != null)
+ {
+ count = ++mapCount;
+
+ if (_logger.isDebugEnabled())
{
- if (_logger.isDebugEnabled())
- {
- _logger.debug("Recording first sighting of JMSMessageID '"
+ msgID + "'");
- }
- _receivedMsgIDs.put(msgID, 1);
+ _logger.debug("Incrementing count for JMSMessageID: '" + msgID
+ "', value now: " + count);
}
+ _receivedMsgIDs.put(msgID, count);
}
- catch(Exception e)
+ else
{
- _logger.warn("Exception recording delivery count for message: " +
msgID, e);
+ count = 1;
+
+ if (_logger.isDebugEnabled())
+ {
+ _logger.debug("Recording first sighting of JMSMessageID '" +
msgID + "'");
+ }
+ _receivedMsgIDs.put(msgID, count);
}
+
+ return count;
}
/**
Modified:
qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate.java
URL:
http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate.java?rev=1049624&r1=1049623&r2=1049624&view=diff
==============================================================================
---
qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate.java
(original)
+++
qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate.java
Wed Dec 15 16:07:08 2010
@@ -134,4 +134,13 @@ public interface AMQMessageDelegate
long getDeliveryTag();
void setJMSMessageID(final UUID messageId) throws JMSException;
+
+ /**
+ * Set the JMSXDeliveryCount property to the given value.
+ *
+ * If the value given is null, the JMSXDeliveryCount property is removed.
+ *
+ * @param count the value to set
+ */
+ void setJMSXDeliveryCount(final Integer count) throws JMSException;
}
Modified:
qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_10.java
URL:
http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_10.java?rev=1049624&r1=1049623&r2=1049624&view=diff
==============================================================================
---
qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_10.java
(original)
+++
qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_10.java
Wed Dec 15 16:07:08 2010
@@ -912,4 +912,18 @@ public class AMQMessageDelegate_0_10 ext
return _deliveryProps;
}
+ public void setJMSXDeliveryCount(Integer count) throws JMSException
+ {
+ String propName = CustomJMSXProperty.JMSXDeliveryCount.toString();
+
+ if(count !=null)
+ {
+ setApplicationHeader(propName, count);
+ }
+ else
+ {
+ removeProperty(propName);
+ }
+ }
+
}
Modified:
qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_8.java
URL:
http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_8.java?rev=1049624&r1=1049623&r2=1049624&view=diff
==============================================================================
---
qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_8.java
(original)
+++
qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_8.java
Wed Dec 15 16:07:08 2010
@@ -574,5 +574,17 @@ public class AMQMessageDelegate_0_8 exte
return _deliveryTag;
}
+ public void setJMSXDeliveryCount(Integer count) throws JMSException
+ {
+ String propName = CustomJMSXProperty.JMSXDeliveryCount.toString();
+ if(count !=null)
+ {
+ getJmsHeaders().setInteger(propName, count);
+ }
+ else
+ {
+ removeProperty(propName);
+ }
+ }
}
Modified:
qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessage.java
URL:
http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessage.java?rev=1049624&r1=1049623&r2=1049624&view=diff
==============================================================================
---
qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessage.java
(original)
+++
qpid/branches/0.5.x-dev/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessage.java
Wed Dec 15 16:07:08 2010
@@ -534,4 +534,8 @@ public abstract class AbstractJMSMessage
_delegate.removeProperty(propertyName);
}
+ public void setJMSXDeliveryCount(Integer count) throws JMSException
+ {
+ _delegate.setJMSXDeliveryCount(count);
+ }
}
Modified:
qpid/branches/0.5.x-dev/qpid/java/client/src/test/java/org/apache/qpid/client/DeliveryCountTrackerTest.java
URL:
http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/java/client/src/test/java/org/apache/qpid/client/DeliveryCountTrackerTest.java?rev=1049624&r1=1049623&r2=1049624&view=diff
==============================================================================
---
qpid/branches/0.5.x-dev/qpid/java/client/src/test/java/org/apache/qpid/client/DeliveryCountTrackerTest.java
(original)
+++
qpid/branches/0.5.x-dev/qpid/java/client/src/test/java/org/apache/qpid/client/DeliveryCountTrackerTest.java
Wed Dec 15 16:07:08 2010
@@ -155,4 +155,29 @@ public class DeliveryCountTrackerTest ex
}
}
}
+
+ /**
+ * Test the count value return as message ID's are tracked
+ */
+ public void testCountReturnedWhenTracking() throws Exception
+ {
+ //check the value is 0 when provided with a null instead of a String id
+ assertEquals("value should have been 0", 0,
_tracker.recordMessage(null, 0));
+
+ //check the value is 0 when provided with a null instead of a String id
+ //and the same delivery tag as before
+ assertEquals("value should have been 0", 0,
_tracker.recordMessage(null, 0));
+
+ //check the value is 1 when provided with an id for the first time
+ for(int id=1 ; id <= CAPACITY; id ++)
+ {
+ assertEquals("value should have been 1", 1,
_tracker.recordMessage(String.valueOf(id), id));
+ }
+
+ //check the value is 2 when provided with an id for the second time
+ for(int id=1 ; id <= CAPACITY; id ++)
+ {
+ assertEquals("value should have been 2", 2,
_tracker.recordMessage(String.valueOf(id), id));
+ }
+ }
}
Modified:
qpid/branches/0.5.x-dev/qpid/java/client/src/test/java/org/apache/qpid/client/message/AbstractJMSMessageTest.java
URL:
http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/java/client/src/test/java/org/apache/qpid/client/message/AbstractJMSMessageTest.java?rev=1049624&r1=1049623&r2=1049624&view=diff
==============================================================================
---
qpid/branches/0.5.x-dev/qpid/java/client/src/test/java/org/apache/qpid/client/message/AbstractJMSMessageTest.java
(original)
+++
qpid/branches/0.5.x-dev/qpid/java/client/src/test/java/org/apache/qpid/client/message/AbstractJMSMessageTest.java
Wed Dec 15 16:07:08 2010
@@ -33,4 +33,53 @@ public class AbstractJMSMessageTest exte
}
}
+ public void testSetJMSXDeliveryCount_0_8() throws JMSException
+ {
+ doTestSetJMSXDeliveryCount(AMQMessageDelegateFactory.FACTORY_0_8);
+ }
+
+ public void testSetJMSXDeliveryCount_0_10() throws JMSException
+ {
+ doTestSetJMSXDeliveryCount(AMQMessageDelegateFactory.FACTORY_0_10);
+ }
+
+ private void doTestSetJMSXDeliveryCount(AMQMessageDelegateFactory<?>
factory) throws JMSException
+ {
+ AbstractJMSMessage abstractMessage = new JMSTextMessage(factory);
+
+ //verify the property is not set
+ assertFalse("property should not yet exist",
abstractMessage.propertyExists("JMSXDeliveryCount"));
+
+ //check that retrieving the property now throws the expected NFE
+ try
+ {
+ abstractMessage.getIntProperty("JMSXDeliveryCount");
+ fail("property should not be set, so NumberFormatException should
be thrown");
+ }
+ catch (NumberFormatException e)
+ {
+ //expected, ignore.
+ }
+
+ //set the value, verify retrieval
+ abstractMessage.setJMSXDeliveryCount(5);
+ assertEquals("Value was incorrect", 5,
abstractMessage.getIntProperty("JMSXDeliveryCount"));
+
+ //remove the property
+ abstractMessage.setJMSXDeliveryCount(null);
+
+ //verify property is cleared
+ assertFalse("property should not yet exist",
abstractMessage.propertyExists("JMSXDeliveryCount"));
+
+ //check that retrieving the property now throws the expected NFE
+ try
+ {
+ abstractMessage.getIntProperty("JMSXDeliveryCount");
+ fail("property should not be set, so NumberFormatException should
be thrown");
+ }
+ catch (NumberFormatException e)
+ {
+ //expected, ignore.
+ }
+ }
}
Modified:
qpid/branches/0.5.x-dev/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/MaxDeliveryCountTest.java
URL:
http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/MaxDeliveryCountTest.java?rev=1049624&r1=1049623&r2=1049624&view=diff
==============================================================================
---
qpid/branches/0.5.x-dev/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/MaxDeliveryCountTest.java
(original)
+++
qpid/branches/0.5.x-dev/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/MaxDeliveryCountTest.java
Wed Dec 15 16:07:08 2010
@@ -42,6 +42,7 @@ import org.apache.qpid.AMQException;
import org.apache.qpid.client.AMQDestination;
import org.apache.qpid.client.AMQQueue;
import org.apache.qpid.client.AMQSession;
+import org.apache.qpid.client.CustomJMSXProperty;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.server.queue.AMQQueueFactory;
import org.apache.qpid.test.utils.QpidTestCase;
@@ -402,6 +403,10 @@ public class MaxDeliveryCountTest extend
}
_deliveryAttempts++; //increment count of times the
current rolled back/recovered message(s) have been seen
+
+ //verify JMSXDeliveryCount
+ assertEquals("Value for JMSXDeliveryCount was not as
expected",
+ _deliveryAttempts,
message.getIntProperty("JMSXDeliveryCount"));
switch(deliveryMode)
{
@@ -541,6 +546,10 @@ public class MaxDeliveryCountTest extend
_deliveryAttempts++; //increment count of times the
current rolled back/recovered message(s) have been seen
+ //verify JMSXDeliveryCount
+ assertEquals("Value for JMSXDeliveryCount was not as
expected",
+ _deliveryAttempts,
message.getIntProperty("JMSXDeliveryCount"));
+
switch(deliveryMode)
{
case Session.SESSION_TRANSACTED:
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:[email protected]
