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 2022/04/25 13:44:52 UTC
[activemq-artemis] branch main updated: ARTEMIS-3798 Allow binary deserialization of message properties AMQ_MSG_MESSAGE_ID, AMQ_MSG_ORIG_DESTINATION, AMQ_MSG_PRODUCER_ID & AMQ_MSG_REPLY_TO from existing messages.
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
The following commit(s) were added to refs/heads/main by this push:
new 83afe0e276 ARTEMIS-3798 Allow binary deserialization of message properties AMQ_MSG_MESSAGE_ID, AMQ_MSG_ORIG_DESTINATION, AMQ_MSG_PRODUCER_ID & AMQ_MSG_REPLY_TO from existing messages.
new 2f72b6068d This closes #4040
83afe0e276 is described below
commit 83afe0e276078d239ad6505ea391ee48350498f6
Author: Marco Roehrs <ma...@twinformatics.at>
AuthorDate: Fri Apr 22 15:35:27 2022 +0200
ARTEMIS-3798 Allow binary deserialization of message properties AMQ_MSG_MESSAGE_ID, AMQ_MSG_ORIG_DESTINATION, AMQ_MSG_PRODUCER_ID & AMQ_MSG_REPLY_TO from existing messages.
co-authored: Justin Bertram contributed OpenWireMessageConverterTest changes
---
.../openwire/OpenWireMessageConverter.java | 47 ++++++++-----
.../openwire/OpenWireMessageConverterTest.java | 76 +++++++++++++++++++++-
2 files changed, 104 insertions(+), 19 deletions(-)
diff --git a/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessageConverter.java b/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessageConverter.java
index a40dab927e..92f6e01360 100644
--- a/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessageConverter.java
+++ b/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessageConverter.java
@@ -94,9 +94,9 @@ public final class OpenWireMessageConverter {
private static final SimpleString AMQ_MSG_COMMAND_ID = new SimpleString(AMQ_PREFIX + "COMMAND_ID");
private static final SimpleString AMQ_MSG_DATASTRUCTURE = new SimpleString(AMQ_PREFIX + "DATASTRUCTURE");
public static final SimpleString AMQ_MSG_MESSAGE_ID = new SimpleString(AMQ_PREFIX + "MESSAGE_ID");
- private static final SimpleString AMQ_MSG_ORIG_DESTINATION = new SimpleString(AMQ_PREFIX + "ORIG_DESTINATION");
+ public static final SimpleString AMQ_MSG_ORIG_DESTINATION = new SimpleString(AMQ_PREFIX + "ORIG_DESTINATION");
public static final SimpleString AMQ_MSG_PRODUCER_ID = new SimpleString(AMQ_PREFIX + "PRODUCER_ID");
- private static final SimpleString AMQ_MSG_REPLY_TO = new SimpleString(AMQ_PREFIX + "REPLY_TO");
+ public static final SimpleString AMQ_MSG_REPLY_TO = new SimpleString(AMQ_PREFIX + "REPLY_TO");
private static final SimpleString AMQ_MSG_USER_ID = new SimpleString(AMQ_PREFIX + "USER_ID");
@@ -622,35 +622,46 @@ public final class OpenWireMessageConverter {
amqMsg.setGroupSequence(coreMessage.getGroupSequence());
- final SimpleString midString = getObjectProperty(coreMessage, SimpleString.class, AMQ_MSG_MESSAGE_ID);
- final MessageId mid;
- if (midString != null) {
- mid = new MessageId(midString.toString());
+ final Object messageIdValue = getObjectProperty(coreMessage, Object.class, AMQ_MSG_MESSAGE_ID);
+ final MessageId messageId;
+ if (messageIdValue instanceof SimpleString) {
+ messageId = new MessageId(messageIdValue.toString());
+ } else if (messageIdValue instanceof byte[]) {
+ ByteSequence midSeq = new ByteSequence((byte[]) messageIdValue);
+ messageId = (MessageId) marshaller.unmarshal(midSeq);
} else {
- //JMSMessageID should be started with "ID:" and needs to be globally unique (node + journal id)
// ARTEMIS-3776 due to AMQ-6431 some older clients will not be able to receive messages
// if using a failover schema due to the messageID overFlowing Integer.MAX_VALUE
String midd = "ID:" + serverNodeUUID + ":-1:-1:" + (coreMessage.getMessageID() / Integer.MAX_VALUE);
- mid = new MessageId(midd, coreMessage.getMessageID() % Integer.MAX_VALUE);
+ messageId = new MessageId(midd, coreMessage.getMessageID() % Integer.MAX_VALUE);
}
- amqMsg.setMessageId(mid);
+ amqMsg.setMessageId(messageId);
- final SimpleString origDestBytes = getObjectProperty(coreMessage, SimpleString.class, AMQ_MSG_ORIG_DESTINATION);
- if (origDestBytes != null) {
- amqMsg.setOriginalDestination(ActiveMQDestination.createDestination(origDestBytes.toString(), QUEUE_TYPE));
+ final Object origDestValue = getObjectProperty(coreMessage, Object.class, AMQ_MSG_ORIG_DESTINATION);
+ if (origDestValue instanceof SimpleString) {
+ amqMsg.setOriginalDestination(ActiveMQDestination.createDestination(origDestValue.toString(), QUEUE_TYPE));
+ } else if (origDestValue instanceof byte[]) {
+ ActiveMQDestination origDest = (ActiveMQDestination) marshaller.unmarshal(new ByteSequence((byte[]) origDestValue));
+ amqMsg.setOriginalDestination(origDest);
}
- final SimpleString producerId = getObjectProperty(coreMessage, SimpleString.class, AMQ_MSG_PRODUCER_ID);
- if (producerId != null && producerId.length() > 0) {
- amqMsg.setProducerId(new ProducerId(producerId.toString()));
+ final Object producerIdValue = getObjectProperty(coreMessage, Object.class, AMQ_MSG_PRODUCER_ID);
+ if (producerIdValue instanceof SimpleString && ((SimpleString) producerIdValue).length() > 0) {
+ amqMsg.setProducerId(new ProducerId(producerIdValue.toString()));
+ } else if (producerIdValue instanceof byte[]) {
+ ProducerId producerId = (ProducerId) marshaller.unmarshal(new ByteSequence((byte[]) producerIdValue));
+ amqMsg.setProducerId(producerId);
}
amqMsg.setRedeliveryCounter(reference.getDeliveryCount() - 1);
- final SimpleString replyToBytes = getObjectProperty(coreMessage, SimpleString.class, AMQ_MSG_REPLY_TO);
- if (replyToBytes != null) {
- amqMsg.setReplyTo(ActiveMQDestination.createDestination(replyToBytes.toString(), QUEUE_TYPE));
+ final Object replyToValue = getObjectProperty(coreMessage, Object.class, AMQ_MSG_REPLY_TO);
+ if (replyToValue instanceof SimpleString) {
+ amqMsg.setReplyTo(ActiveMQDestination.createDestination(replyToValue.toString(), QUEUE_TYPE));
+ } else if (replyToValue instanceof byte[]) {
+ ActiveMQDestination replyTo = (ActiveMQDestination) marshaller.unmarshal(new ByteSequence((byte[]) replyToValue));
+ amqMsg.setReplyTo(replyTo);
}
final SimpleString userId = getObjectProperty(coreMessage, SimpleString.class, AMQ_MSG_USER_ID);
diff --git a/artemis-protocols/artemis-openwire-protocol/src/test/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessageConverterTest.java b/artemis-protocols/artemis-openwire-protocol/src/test/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessageConverterTest.java
index ce70a1d3f2..756f8a6825 100644
--- a/artemis-protocols/artemis-openwire-protocol/src/test/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessageConverterTest.java
+++ b/artemis-protocols/artemis-openwire-protocol/src/test/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessageConverterTest.java
@@ -25,6 +25,7 @@ import org.apache.activemq.artemis.core.server.MessageReference;
import org.apache.activemq.artemis.core.server.Queue;
import org.apache.activemq.artemis.core.server.impl.MessageReferenceImpl;
import org.apache.activemq.artemis.reader.MessageUtil;
+import org.apache.activemq.artemis.utils.RandomUtil;
import org.apache.activemq.artemis.utils.UUID;
import org.apache.activemq.artemis.utils.UUIDGenerator;
import org.apache.activemq.command.ActiveMQDestination;
@@ -34,10 +35,15 @@ import org.apache.activemq.command.MessageDispatch;
import org.apache.activemq.command.MessageId;
import org.apache.activemq.command.ProducerId;
import org.apache.activemq.openwire.OpenWireFormatFactory;
+import org.apache.activemq.util.ByteSequence;
import org.apache.activemq.wireformat.WireFormat;
import org.junit.Test;
import org.mockito.Mockito;
+import static org.apache.activemq.artemis.core.protocol.openwire.OpenWireMessageConverter.AMQ_MSG_MESSAGE_ID;
+import static org.apache.activemq.artemis.core.protocol.openwire.OpenWireMessageConverter.AMQ_MSG_ORIG_DESTINATION;
+import static org.apache.activemq.artemis.core.protocol.openwire.OpenWireMessageConverter.AMQ_MSG_PRODUCER_ID;
+import static org.apache.activemq.artemis.core.protocol.openwire.OpenWireMessageConverter.AMQ_MSG_REPLY_TO;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
@@ -150,6 +156,23 @@ public class OpenWireMessageConverterTest {
assertEquals(PRODUCER_ID, classicMessageDispatch.getMessage().getProducerId().toString());
}
+ @Test
+ public void testLegacyProducerId() throws Exception {
+ final String PRODUCER_ID = "123:456:789";
+
+ ActiveMQMessage classicMessage = new ActiveMQMessage();
+ classicMessage.setProducerId(new ProducerId(PRODUCER_ID));
+ final ByteSequence pidBytes = openWireFormat.marshal(classicMessage.getProducerId());
+ pidBytes.compact();
+ ICoreMessage coreMessage = new CoreMessage().initBuffer(8);
+ coreMessage.putBytesProperty(AMQ_MSG_PRODUCER_ID, pidBytes.data);
+ MessageReference messageReference = new MessageReferenceImpl(coreMessage, Mockito.mock(Queue.class));
+ AMQConsumer amqConsumer = Mockito.mock(AMQConsumer.class);
+ Mockito.when(amqConsumer.getOpenwireDestination()).thenReturn(destination);
+ MessageDispatch messageDispatch = OpenWireMessageConverter.createMessageDispatch(messageReference, coreMessage, openWireFormat, amqConsumer, nodeUUID);
+ assertEquals(PRODUCER_ID, messageDispatch.getMessage().getProducerId().toString());
+ }
+
@Test
public void testMessageId() throws Exception {
final String MESSAGE_ID = "ID:123:456:789";
@@ -157,7 +180,7 @@ public class OpenWireMessageConverterTest {
ActiveMQMessage classicMessage = new ActiveMQMessage();
classicMessage.setMessageId(new MessageId(MESSAGE_ID));
Message artemisMessage = OpenWireMessageConverter.inbound(classicMessage.getMessage(), openWireFormat, null);
- assertEquals(MESSAGE_ID, artemisMessage.getStringProperty(OpenWireMessageConverter.AMQ_MSG_MESSAGE_ID));
+ assertEquals(MESSAGE_ID, artemisMessage.getStringProperty(AMQ_MSG_MESSAGE_ID));
MessageReference messageReference = new MessageReferenceImpl(artemisMessage, Mockito.mock(Queue.class));
AMQConsumer amqConsumer = Mockito.mock(AMQConsumer.class);
@@ -166,6 +189,57 @@ public class OpenWireMessageConverterTest {
assertEquals(MESSAGE_ID, classicMessageDispatch.getMessage().getMessageId().toString());
}
+ @Test
+ public void testLegacyMessageId() throws Exception {
+ final String MESSAGE_ID = "ID:123:456:789";
+
+ ActiveMQMessage classicMessage = new ActiveMQMessage();
+ classicMessage.setMessageId(new MessageId(MESSAGE_ID));
+ final ByteSequence midBytes = openWireFormat.marshal(classicMessage.getMessageId());
+ midBytes.compact();
+ ICoreMessage coreMessage = new CoreMessage().initBuffer(8);
+ coreMessage.putBytesProperty(AMQ_MSG_MESSAGE_ID, midBytes.data);
+ MessageReference messageReference = new MessageReferenceImpl(coreMessage, Mockito.mock(Queue.class));
+ AMQConsumer amqConsumer = Mockito.mock(AMQConsumer.class);
+ Mockito.when(amqConsumer.getOpenwireDestination()).thenReturn(destination);
+ MessageDispatch messageDispatch = OpenWireMessageConverter.createMessageDispatch(messageReference, coreMessage, openWireFormat, amqConsumer, nodeUUID);
+ assertEquals(MESSAGE_ID, messageDispatch.getMessage().getMessageId().toString());
+ }
+
+ @Test
+ public void testLegacyOriginalDestination() throws Exception {
+ final String DESTINATION = RandomUtil.randomString().replace("-", "");
+
+ ActiveMQMessage classicMessage = new ActiveMQMessage();
+ classicMessage.setOriginalDestination(new ActiveMQQueue(DESTINATION));
+ final ByteSequence destBytes = openWireFormat.marshal(classicMessage.getOriginalDestination());
+ destBytes.compact();
+ ICoreMessage coreMessage = new CoreMessage().initBuffer(8);
+ coreMessage.putBytesProperty(AMQ_MSG_ORIG_DESTINATION, destBytes.data);
+ MessageReference messageReference = new MessageReferenceImpl(coreMessage, Mockito.mock(Queue.class));
+ AMQConsumer amqConsumer = Mockito.mock(AMQConsumer.class);
+ Mockito.when(amqConsumer.getOpenwireDestination()).thenReturn(destination);
+ MessageDispatch messageDispatch = OpenWireMessageConverter.createMessageDispatch(messageReference, coreMessage, openWireFormat, amqConsumer, nodeUUID);
+ assertEquals("queue://" + DESTINATION, messageDispatch.getMessage().getOriginalDestination().toString());
+ }
+
+ @Test
+ public void testLegacyReplyTo() throws Exception {
+ final String DESTINATION = RandomUtil.randomString().replace("-", "");
+
+ ActiveMQMessage classicMessage = new ActiveMQMessage();
+ classicMessage.setJMSReplyTo(new ActiveMQQueue(DESTINATION));
+ final ByteSequence destBytes = openWireFormat.marshal(classicMessage.getJMSReplyTo());
+ destBytes.compact();
+ ICoreMessage coreMessage = new CoreMessage().initBuffer(8);
+ coreMessage.putBytesProperty(AMQ_MSG_REPLY_TO, destBytes.data);
+ MessageReference messageReference = new MessageReferenceImpl(coreMessage, Mockito.mock(Queue.class));
+ AMQConsumer amqConsumer = Mockito.mock(AMQConsumer.class);
+ Mockito.when(amqConsumer.getOpenwireDestination()).thenReturn(destination);
+ MessageDispatch messageDispatch = OpenWireMessageConverter.createMessageDispatch(messageReference, coreMessage, openWireFormat, amqConsumer, nodeUUID);
+ assertEquals("queue://" + DESTINATION, messageDispatch.getMessage().getReplyTo().toString());
+ }
+
@Test
public void testBadPropertyConversion() throws Exception {
final String hdrArrival = "__HDR_ARRIVAL";