QPID-315

Moved message conversion logic from BasicMessageProducer to MessageConverter
Added correlation id to AbstractJMSMessage.toString()

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

3 files changed, 12 insertions, 99 deletions

diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java b/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java
index 8bc98bc495..04df584a8e 100644
--- a/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java
+++ b/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java
@@ -25,6 +25,7 @@ import org.apache.mina.common.ByteBuffer;
 import org.apache.qpid.AMQException;
 import org.apache.qpid.client.message.AbstractJMSMessage;
 import org.apache.qpid.client.message.JMSBytesMessage;
+import org.apache.qpid.client.message.MessageConverter;
 import org.apache.qpid.client.protocol.AMQProtocolHandler;
 import org.apache.qpid.framing.*;
 
@@ -367,112 +368,31 @@ public class BasicMessageProducer extends Closeable implements org.apache.qpid.j
 
             if (message instanceof BytesMessage)
             {
-                BytesMessage bytesMessage = (BytesMessage) message;
-                bytesMessage.reset();
-
-                JMSBytesMessage nativeMsg = (JMSBytesMessage) _session.createBytesMessage();
-
-
-                byte[] buf = new byte[1024];
-
-                int len;
-
-                while ((len = bytesMessage.readBytes(buf)) != -1)
-                {
-                    nativeMsg.writeBytes(buf, 0, len);
-                }
-
-                newMessage = nativeMsg;
+                newMessage = new MessageConverter((BytesMessage)message).getConvertedMessage();
             }
             else if (message instanceof MapMessage)
             {
-                MapMessage origMessage = (MapMessage) message;
-                MapMessage nativeMessage = _session.createMapMessage();
-
-                Enumeration mapNames = origMessage.getMapNames();
-                while (mapNames.hasMoreElements())
-                {
-                    String name = (String) mapNames.nextElement();
-                    nativeMessage.setObject(name, origMessage.getObject(name));
-                }
-                newMessage = (AbstractJMSMessage) nativeMessage;
+                newMessage = new MessageConverter((MapMessage)message).getConvertedMessage();
             }
             else if (message instanceof ObjectMessage)
             {
-                ObjectMessage origMessage = (ObjectMessage) message;
-                ObjectMessage nativeMessage = _session.createObjectMessage();
-
-                nativeMessage.setObject(origMessage.getObject());
-
-                newMessage = (AbstractJMSMessage) nativeMessage;
+                newMessage = new MessageConverter((ObjectMessage)message).getConvertedMessage();
             }
             else if (message instanceof TextMessage)
             {
-                TextMessage origMessage = (TextMessage) message;
-                TextMessage nativeMessage = _session.createTextMessage();
-
-                nativeMessage.setText(origMessage.getText());
-
-                newMessage = (AbstractJMSMessage) nativeMessage;
+               newMessage = new MessageConverter((TextMessage)message).getConvertedMessage();
             }
             else if (message instanceof StreamMessage)
             {
-                StreamMessage origMessage = (StreamMessage) message;
-                StreamMessage nativeMessage = _session.createStreamMessage();
-
-
-                try
-                {
-                    origMessage.reset();
-                    while (true)
-                    {
-                        nativeMessage.writeObject(origMessage.readObject());
-                    }
-                }
-                catch (MessageEOFException e)
-                {
-                    ;//
-                }
-                newMessage = (AbstractJMSMessage) nativeMessage;
+                newMessage = new MessageConverter((StreamMessage)message).getConvertedMessage();
             }
             else
             {
+                //TODO; Do we really want to create an empty message here ?
                 newMessage = (AbstractJMSMessage) _session.createMessage();
-
+                return new MessageConverter(newMessage).getConvertedMessage();
             }
 
-            Enumeration propertyNames = message.getPropertyNames();
-            while (propertyNames.hasMoreElements())
-            {
-                String propertyName = String.valueOf(propertyNames.nextElement());
-                if (!propertyName.startsWith("JMSX_"))
-                {
-                    Object value = message.getObjectProperty(propertyName);
-                    newMessage.setObjectProperty(propertyName, value);
-                }
-            }
-
-            newMessage.setJMSDeliveryMode(message.getJMSDeliveryMode());
-
-
-            int priority = message.getJMSPriority();
-            if (priority < 0)
-            {
-                priority = 0;
-            }
-            else if (priority > 9)
-            {
-                priority = 9;
-            }
-
-            newMessage.setJMSPriority(priority);
-            if (message.getJMSReplyTo() != null)
-            {
-                newMessage.setJMSReplyTo(message.getJMSReplyTo());
-            }
-            newMessage.setJMSType(message.getJMSType());
-
-
             if (newMessage != null)
             {
                 return newMessage;
diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessage.java b/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessage.java
index da8d388088..9dc4339895 100644
--- a/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessage.java
+++ b/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessage.java
@@ -471,6 +471,7 @@ public abstract class AbstractJMSMessage extends AMQMessage implements org.apach
         {
             StringBuffer buf = new StringBuffer("Body:\n");
             buf.append(toBodyString());
+            buf.append("\nJMS Correlation ID: ").append(getJMSCorrelationID());
             buf.append("\nJMS timestamp: ").append(getJMSTimestamp());
             buf.append("\nJMS expiration: ").append(getJMSExpiration());
             buf.append("\nJMS priority: ").append(getJMSPriority());
diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/message/MessageConverter.java b/qpid/java/client/src/main/java/org/apache/qpid/client/message/MessageConverter.java
index dcf08234c5..58089f595b 100644
--- a/qpid/java/client/src/main/java/org/apache/qpid/client/message/MessageConverter.java
+++ b/qpid/java/client/src/main/java/org/apache/qpid/client/message/MessageConverter.java
@@ -37,17 +37,9 @@ public class MessageConverter {
      */
     private AbstractJMSMessage _newMessage;
 
-    /*
-    * Message instance is not one of the Qpid implementations
-    */
-    public MessageConverter(Message message)
+    public MessageConverter(AbstractJMSMessage message) throws JMSException
     {
-        _newMessage = new JMSBytesMessage();
-    }
-
-    public MessageConverter(AbstractJMSMessage message)
-    {
-
+        _newMessage = message;
     }
 
     public MessageConverter(BytesMessage message) throws JMSException
@@ -149,7 +141,7 @@ public class MessageConverter {
         while (propertyNames.hasMoreElements())
         {
             String propertyName = String.valueOf(propertyNames.nextElement());
-            //TODO: Shouldn't need to check for JMS properties here as don't think getPropertyNames() returns them
+            //TODO: Shouldn't need to check for JMS properties here as don't think getPropertyNames() should return them
             if (!propertyName.startsWith("JMSX_"))
             {
                 Object value = message.getObjectProperty(propertyName);