QPID-6047 : [Java Broker] Improve Message conversion between 1.0 and 0.x protocols - add mapping from application property qpid.subject to the 1.0 subject property

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

4 files changed, 45 insertions, 9 deletions

diff --git a/qpid/java/broker-plugins/amqp-msg-conv-0-10-to-1-0/src/main/java/org/apache/qpid/server/protocol/converter/v0_10_v1_0/MessageConverter_0_10_to_1_0.java b/qpid/java/broker-plugins/amqp-msg-conv-0-10-to-1-0/src/main/java/org/apache/qpid/server/protocol/converter/v0_10_v1_0/MessageConverter_0_10_to_1_0.java
index 84a57d474a..506e6b1850 100644
--- a/qpid/java/broker-plugins/amqp-msg-conv-0-10-to-1-0/src/main/java/org/apache/qpid/server/protocol/converter/v0_10_v1_0/MessageConverter_0_10_to_1_0.java
+++ b/qpid/java/broker-plugins/amqp-msg-conv-0-10-to-1-0/src/main/java/org/apache/qpid/server/protocol/converter/v0_10_v1_0/MessageConverter_0_10_to_1_0.java
@@ -22,7 +22,10 @@
 package org.apache.qpid.server.protocol.converter.v0_10_v1_0;
 
 import java.util.ArrayList;
+import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Map;
+
 import org.apache.qpid.amqp_1_0.messaging.SectionEncoder;
 import org.apache.qpid.amqp_1_0.type.Binary;
 import org.apache.qpid.amqp_1_0.type.Section;
@@ -126,9 +129,18 @@ public class MessageConverter_0_10_to_1_0  extends MessageConverter_to_1_0<Messa
 
             sections.add(props);
 
-            if(msgProps.getApplicationHeaders() != null)
+            Map<String, Object> applicationProperties = msgProps.getApplicationHeaders();
+
+            if(applicationProperties != null)
             {
-                sections.add(new ApplicationProperties(msgProps.getApplicationHeaders()));
+                if(applicationProperties.containsKey("qpid.subject"))
+                {
+                    props.setSubject(String.valueOf(applicationProperties.get("qpid.subject")));
+                    applicationProperties = new LinkedHashMap<>(applicationProperties);
+                    applicationProperties.remove("qpid.subject");
+                }
+                sections.add(new ApplicationProperties(applicationProperties));
+
             }
         }
         return new MessageMetaData_1_0(sections, sectionEncoder);
diff --git a/qpid/java/broker-plugins/amqp-msg-conv-0-10-to-1-0/src/main/java/org/apache/qpid/server/protocol/converter/v0_10_v1_0/MessageConverter_1_0_to_v0_10.java b/qpid/java/broker-plugins/amqp-msg-conv-0-10-to-1-0/src/main/java/org/apache/qpid/server/protocol/converter/v0_10_v1_0/MessageConverter_1_0_to_v0_10.java
index 54d4638bb8..0c47ddf137 100644
--- a/qpid/java/broker-plugins/amqp-msg-conv-0-10-to-1-0/src/main/java/org/apache/qpid/server/protocol/converter/v0_10_v1_0/MessageConverter_1_0_to_v0_10.java
+++ b/qpid/java/broker-plugins/amqp-msg-conv-0-10-to-1-0/src/main/java/org/apache/qpid/server/protocol/converter/v0_10_v1_0/MessageConverter_1_0_to_v0_10.java
@@ -21,14 +21,15 @@
 package org.apache.qpid.server.protocol.converter.v0_10_v1_0;
 
 import java.nio.ByteBuffer;
+import java.util.LinkedHashMap;
 import java.util.Map;
 
-import org.apache.qpid.server.message.AMQMessageHeader;
 import org.apache.qpid.server.plugin.MessageConverter;
 import org.apache.qpid.server.plugin.PluggableService;
 import org.apache.qpid.server.protocol.v0_10.MessageMetaData_0_10;
 import org.apache.qpid.server.protocol.v0_10.MessageTransferMessage;
 import org.apache.qpid.server.protocol.v1_0.MessageConverter_from_1_0;
+import org.apache.qpid.server.protocol.v1_0.MessageMetaData_1_0;
 import org.apache.qpid.server.protocol.v1_0.Message_1_0;
 import org.apache.qpid.server.store.StoredMessage;
 import org.apache.qpid.server.virtualhost.VirtualHostImpl;
@@ -138,7 +139,7 @@ public class MessageConverter_1_0_to_v0_10 implements MessageConverter<Message_1
         DeliveryProperties deliveryProps = new DeliveryProperties();
         MessageProperties messageProps = new MessageProperties();
 
-        final AMQMessageHeader origHeader = serverMsg.getMessageHeader();
+        final MessageMetaData_1_0.MessageHeader_1_0  origHeader = serverMsg.getMessageHeader();
 
 
         deliveryProps.setExpiration(serverMsg.getExpiration());
@@ -177,8 +178,15 @@ public class MessageConverter_1_0_to_v0_10 implements MessageConverter<Message_1
             messageProps.setReplyTo(replyTo);
         }
 
-        messageProps.setApplicationHeaders((Map<String, Object>) MessageConverter_from_1_0.convertValue(serverMsg.getMessageHeader()
-                                                                                                                .getHeadersAsMap()));
+        Map<String, Object> appHeaders =
+                (Map<String, Object>) MessageConverter_from_1_0.convertValue(serverMsg.getMessageHeader()
+                                                                                     .getHeadersAsMap());
+        if(origHeader.getSubject() != null && !appHeaders.containsKey("qpid.subject"))
+        {
+            appHeaders = new LinkedHashMap<>(appHeaders);
+            appHeaders.put("qpid.subject", origHeader.getSubject());
+        }
+        messageProps.setApplicationHeaders(appHeaders);
 
         Header header = new Header(deliveryProps, messageProps, null);
         return new MessageMetaData_0_10(header, size, serverMsg.getArrivalTime());
diff --git a/qpid/java/broker-plugins/amqp-msg-conv-0-8-to-1-0/src/main/java/org/apache/qpid/server/protocol/converter/v0_8_v1_0/MessageConverter_0_8_to_1_0.java b/qpid/java/broker-plugins/amqp-msg-conv-0-8-to-1-0/src/main/java/org/apache/qpid/server/protocol/converter/v0_8_v1_0/MessageConverter_0_8_to_1_0.java
index eb9d6842e4..553b1ba662 100644
--- a/qpid/java/broker-plugins/amqp-msg-conv-0-8-to-1-0/src/main/java/org/apache/qpid/server/protocol/converter/v0_8_v1_0/MessageConverter_0_8_to_1_0.java
+++ b/qpid/java/broker-plugins/amqp-msg-conv-0-8-to-1-0/src/main/java/org/apache/qpid/server/protocol/converter/v0_8_v1_0/MessageConverter_0_8_to_1_0.java
@@ -22,7 +22,10 @@ package org.apache.qpid.server.protocol.converter.v0_8_v1_0;
 
 import java.net.URISyntaxException;
 import java.util.ArrayList;
+import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Map;
+
 import org.apache.qpid.amqp_1_0.messaging.SectionEncoder;
 import org.apache.qpid.amqp_1_0.type.Binary;
 import org.apache.qpid.amqp_1_0.type.Section;
@@ -152,7 +155,15 @@ public class MessageConverter_0_8_to_1_0 extends MessageConverter_to_1_0<AMQMess
 
         sections.add(props);
 
-        sections.add(new ApplicationProperties(FieldTable.convertToMap(contentHeader.getHeaders())));
+        Map<String, Object> applicationProperties = FieldTable.convertToMap(contentHeader.getHeaders());
+
+        if(applicationProperties.containsKey("qpid.subject"))
+        {
+            props.setSubject(String.valueOf(applicationProperties.get("qpid.subject")));
+            applicationProperties = new LinkedHashMap<>(applicationProperties);
+            applicationProperties.remove("qpid.subject");
+        }
+        sections.add(new ApplicationProperties(applicationProperties));
 
         return new MessageMetaData_1_0(sections, sectionEncoder);
     }
diff --git a/qpid/java/broker-plugins/amqp-msg-conv-0-8-to-1-0/src/main/java/org/apache/qpid/server/protocol/converter/v0_8_v1_0/MessageConverter_1_0_to_v0_8.java b/qpid/java/broker-plugins/amqp-msg-conv-0-8-to-1-0/src/main/java/org/apache/qpid/server/protocol/converter/v0_8_v1_0/MessageConverter_1_0_to_v0_8.java
index 2de21e1a9f..783c6ca772 100644
--- a/qpid/java/broker-plugins/amqp-msg-conv-0-8-to-1-0/src/main/java/org/apache/qpid/server/protocol/converter/v0_8_v1_0/MessageConverter_1_0_to_v0_8.java
+++ b/qpid/java/broker-plugins/amqp-msg-conv-0-8-to-1-0/src/main/java/org/apache/qpid/server/protocol/converter/v0_8_v1_0/MessageConverter_1_0_to_v0_8.java
@@ -136,6 +136,8 @@ public class MessageConverter_1_0_to_v0_8 implements MessageConverter<Message_1_
     private MessageMetaData convertMetaData(final Message_1_0 serverMsg, final String bodyMimeType, final int size)
     {
 
+        final MessageMetaData_1_0.MessageHeader_1_0 header = serverMsg.getMessageHeader();
+
         MessagePublishInfo publishInfo = new MessagePublishInfo()
                                             {
                                                 @Override
@@ -165,8 +167,6 @@ public class MessageConverter_1_0_to_v0_8 implements MessageConverter<Message_1_
                                                 @Override
                                                 public AMQShortString getRoutingKey()
                                                 {
-                                                    MessageMetaData_1_0.MessageHeader_1_0 header =
-                                                            serverMsg.getMessageHeader();
                                                     String key = header.getTo();
                                                     if(key == null)
                                                     {
@@ -192,6 +192,11 @@ public class MessageConverter_1_0_to_v0_8 implements MessageConverter<Message_1_
 
         Map<String,Object> headerProps = new LinkedHashMap<String, Object>();
 
+        if(header.getSubject() != null)
+        {
+            headerProps.put("qpid.subject", header.getSubject());
+        }
+
         for(String headerName : serverMsg.getMessageHeader().getHeaderNames())
         {
             headerProps.put(headerName, MessageConverter_from_1_0.convertValue(serverMsg.getMessageHeader().getHeader(headerName)));