You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cl...@apache.org on 2017/11/27 20:04:32 UTC
[2/3] activemq-artemis git commit: ARTEMIS-1508 Fixing Divert message
and AMQP.reencode
ARTEMIS-1508 Fixing Divert message and AMQP.reencode
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/172420c5
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/172420c5
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/172420c5
Branch: refs/heads/master
Commit: 172420c5531d6865360279b3e1456d452b6338ec
Parents: 9a6649d
Author: Clebert Suconic <cl...@apache.org>
Authored: Mon Nov 27 11:31:10 2017 -0500
Committer: Clebert Suconic <cl...@apache.org>
Committed: Mon Nov 27 15:04:11 2017 -0500
----------------------------------------------------------------------
.../protocol/amqp/broker/AMQPMessage.java | 10 +++-
.../protocol/amqp/message/AMQPMessageTest.java | 48 +++++++++++++++++++-
2 files changed, 55 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/172420c5/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessage.java
----------------------------------------------------------------------
diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessage.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessage.java
index a03ff10..06a5894 100644
--- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessage.java
+++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessage.java
@@ -966,10 +966,18 @@ public class AMQPMessage extends RefCountMessage {
@Override
public void reencode() {
+ parseHeaders();
+ getApplicationProperties();
+ if (_header != null) getProtonMessage().setHeader(_header);
if (_deliveryAnnotations != null) getProtonMessage().setDeliveryAnnotations(_deliveryAnnotations);
if (_messageAnnotations != null) getProtonMessage().setMessageAnnotations(_messageAnnotations);
if (applicationProperties != null) getProtonMessage().setApplicationProperties(applicationProperties);
- if (_properties != null) getProtonMessage().setProperties(this._properties);
+ if (_properties != null) {
+ if (address != null) {
+ _properties.setTo(address);
+ }
+ getProtonMessage().setProperties(this._properties);
+ }
bufferValid = false;
checkBuffer();
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/172420c5/artemis-protocols/artemis-amqp-protocol/src/test/java/org/apache/activemq/artemis/protocol/amqp/message/AMQPMessageTest.java
----------------------------------------------------------------------
diff --git a/artemis-protocols/artemis-amqp-protocol/src/test/java/org/apache/activemq/artemis/protocol/amqp/message/AMQPMessageTest.java b/artemis-protocols/artemis-amqp-protocol/src/test/java/org/apache/activemq/artemis/protocol/amqp/message/AMQPMessageTest.java
index 9dcf7c9..bff43a8 100644
--- a/artemis-protocols/artemis-amqp-protocol/src/test/java/org/apache/activemq/artemis/protocol/amqp/message/AMQPMessageTest.java
+++ b/artemis-protocols/artemis-amqp-protocol/src/test/java/org/apache/activemq/artemis/protocol/amqp/message/AMQPMessageTest.java
@@ -24,6 +24,7 @@ import static org.junit.Assert.assertTrue;
import java.nio.charset.StandardCharsets;
import java.util.Date;
+import java.util.HashMap;
import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.ActiveMQBuffers;
@@ -34,7 +35,6 @@ import org.apache.activemq.artemis.protocol.amqp.broker.AMQPMessagePersisterV2;
import org.apache.activemq.artemis.protocol.amqp.util.NettyWritable;
import org.apache.activemq.artemis.spi.core.protocol.EmbedMessageUtil;
import org.apache.activemq.artemis.utils.RandomUtil;
-import org.apache.commons.collections.map.HashedMap;
import org.apache.qpid.proton.amqp.UnsignedInteger;
import org.apache.qpid.proton.amqp.messaging.ApplicationProperties;
import org.apache.qpid.proton.amqp.messaging.Header;
@@ -58,7 +58,7 @@ public class AMQPMessageTest {
protonMessage.setProperties(properties);
protonMessage.getHeader().setDeliveryCount(new UnsignedInteger(7));
protonMessage.getHeader().setDurable(Boolean.TRUE);
- protonMessage.setApplicationProperties(new ApplicationProperties(new HashedMap()));
+ protonMessage.setApplicationProperties(new ApplicationProperties(new HashMap()));
AMQPMessage decoded = encodeAndDecodeMessage(protonMessage);
@@ -68,6 +68,39 @@ public class AMQPMessageTest {
}
@Test
+ public void testApplicationPropertiesReencode() {
+ MessageImpl protonMessage = (MessageImpl) Message.Factory.create();
+ protonMessage.setHeader( new Header());
+ Properties properties = new Properties();
+ properties.setTo("someNiceLocal");
+ protonMessage.setProperties(properties);
+ protonMessage.getHeader().setDeliveryCount(new UnsignedInteger(7));
+ protonMessage.getHeader().setDurable(Boolean.TRUE);
+ HashMap map = new HashMap();
+ map.put("key", "string1");
+ protonMessage.setApplicationProperties(new ApplicationProperties(map));
+
+ AMQPMessage decoded = encodeAndDecodeMessage(protonMessage);
+ assertEquals("someNiceLocal", decoded.getAddress());
+
+ decoded.setAddress("newAddress");
+
+ decoded.reencode();
+ assertEquals(7, decoded.getHeader().getDeliveryCount().intValue());
+ assertEquals(true, decoded.getHeader().getDurable());
+ assertEquals("newAddress", decoded.getAddress());
+ assertEquals("string1", decoded.getObjectProperty("key"));
+
+ // validate if the message will be the same after delivery
+ AMQPMessage newDecoded = encodeDelivery(decoded, 3);
+ assertEquals(2, decoded.getHeader().getDeliveryCount().intValue());
+ assertEquals(true, newDecoded.getHeader().getDurable());
+ assertEquals("newAddress", newDecoded.getAddress());
+ assertEquals("string1", newDecoded.getObjectProperty("key"));
+
+ }
+
+ @Test
public void testGetAddressFromMessage() {
final String ADDRESS = "myQueue";
@@ -260,4 +293,15 @@ public class AMQPMessageTest {
return new AMQPMessage(0, bytes);
}
+
+ private AMQPMessage encodeDelivery(AMQPMessage message, int deliveryCount) {
+ ByteBuf nettyBuffer = Unpooled.buffer(1500);
+
+ message.sendBuffer(nettyBuffer, deliveryCount);
+
+ byte[] bytes = new byte[nettyBuffer.writerIndex()];
+ nettyBuffer.readBytes(bytes);
+
+ return new AMQPMessage(0, bytes);
+ }
}