Apache
/
activemq-artemis
mirror of https://github.com/apache/activemq-artemis.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

ARTEMIS-1508 Fixing Divert message and AMQP.reencode

This commit is contained in:
Clebert Suconic 2017-11-27 11:31:10 -05:00
parent 9a6649d753
commit 172420c553
2 changed files with 55 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessage.java
View File

@ -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();
}

48
artemis-protocols/artemis-amqp-protocol/src/test/java/org/apache/activemq/artemis/protocol/amqp/message/AMQPMessageTest.java
View File

@ -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);
@ -67,6 +67,39 @@ public class AMQPMessageTest {
assertEquals("someNiceLocal", decoded.getAddress());
}
@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);
}
}