QPID-2242 : Update to 0-8 producer to set the JMS_QPID_DESTTYPE when sending messages.

This update required a change to the FieldTable to clear any _encodedBuffer when setting a new value
Provided test in JMSDestintationTest. This test requires a broker that supports both 0-8/9 and 0-10 as we must first use a 0-10 connection to send a message, which does not set JMS_QPID_DESTTYPE, then receive the message on 0-8/9. This allows us to validate that messages recieved without JMS_QPID_DESTTYPE set will can correctly set the value without a BufferOverflow Exception when simply forwarding the message.

Excluded test from all cpp and InVM runs as test requires a 0-8/9 and 0-10 capable broker. Can reinstate to InVM when multiprotocol testing is capable.

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@888345 13f79535-47bb-0310-9956-ffa450edef68

1 files changed, 21 insertions, 1 deletions

diff --git a/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_8.java b/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_8.java
index 048065eac9..f726cd02a2 100644
--- a/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_8.java
+++ b/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_8.java
@@ -24,6 +24,8 @@ import java.util.UUID;
 
 import javax.jms.JMSException;
 import javax.jms.Message;
+import javax.jms.Topic;
+import javax.jms.Queue;
 
 import org.apache.mina.common.ByteBuffer;
 import org.apache.qpid.client.message.AbstractJMSMessage;
@@ -87,7 +89,25 @@ public class BasicMessageProducer_0_8 extends BasicMessageProducer
         BasicContentHeaderProperties contentHeaderProperties = delegate.getContentHeaderProperties();
 
         contentHeaderProperties.setUserId(_userID);
-        
+
+        //Set the JMS_QPID_DESTTYPE for 0-8/9 messages
+        int type;
+        if (destination instanceof Topic)
+        {
+            type = AMQDestination.TOPIC_TYPE;
+        }
+        else if (destination instanceof Queue)
+        {
+            type = AMQDestination.QUEUE_TYPE;
+        }
+        else
+        {
+            type = AMQDestination.UNKNOWN_TYPE;
+        }
+
+        //Set JMS_QPID_DESTTYPE
+        delegate.getContentHeaderProperties().getHeaders().setInteger(CustomJMSXProperty.JMS_QPID_DESTTYPE.getShortStringName(), type);
+
         if (!_disableTimestamps)
         {
             final long currentTime = System.currentTimeMillis();