You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by or...@apache.org on 2017/08/01 16:09:40 UTC
[1/3] qpid-broker-j git commit: QPID-7434: Add unit tests for
property conversion of internal messages to AMQP 1.0 messages
Repository: qpid-broker-j
Updated Branches:
refs/heads/master 4a2d88957 -> 8ab67e0c1
QPID-7434: Add unit tests for property conversion of internal messages to AMQP 1.0 messages
Project: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/commit/04b19e96
Tree: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/tree/04b19e96
Diff: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/diff/04b19e96
Branch: refs/heads/master
Commit: 04b19e96b30e92d2270458c86381d356726cc76d
Parents: 4a2d889
Author: Alex Rudyy <or...@apache.org>
Authored: Tue Aug 1 15:30:20 2017 +0100
Committer: Alex Rudyy <or...@apache.org>
Committed: Tue Aug 1 15:30:20 2017 +0100
----------------------------------------------------------------------
.../message/internal/InternalMessage.java | 30 +-
.../v1_0/MessageConverter_Internal_to_v1_0.java | 57 +++-
.../v1_0/MessageConverter_from_1_0.java | 30 ++
.../PropertyConverter_Internal_to_v1_0Test.java | 307 +++++++++++++++++++
4 files changed, 410 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/04b19e96/broker-core/src/main/java/org/apache/qpid/server/message/internal/InternalMessage.java
----------------------------------------------------------------------
diff --git a/broker-core/src/main/java/org/apache/qpid/server/message/internal/InternalMessage.java b/broker-core/src/main/java/org/apache/qpid/server/message/internal/InternalMessage.java
index 102bf09..a65b58c 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/message/internal/InternalMessage.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/message/internal/InternalMessage.java
@@ -33,6 +33,7 @@ import java.util.List;
import java.util.Map;
import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.server.bytebuffer.QpidByteBufferUtils;
import org.apache.qpid.server.message.AMQMessageHeader;
import org.apache.qpid.server.message.AbstractServerMessageImpl;
import org.apache.qpid.server.store.MessageHandle;
@@ -68,21 +69,28 @@ public class InternalMessage extends AbstractServerMessageImpl<InternalMessage,
{
super(msg, null);
_header = msg.getMetaData().getHeader();
- Collection<QpidByteBuffer> bufs = msg.getContent(0, (int)getSize());
-
- try(ObjectInputStream is = new ObjectInputStream(new ByteBufferInputStream(ByteBufferUtils.combine(bufs))))
+ long contentSize = getSize();
+ if (contentSize > 0)
{
- _messageBody = is.readObject();
+ Collection<QpidByteBuffer> bufs = msg.getContent(0, (int) contentSize);
+ try (ObjectInputStream is = new ObjectInputStream(new ByteBufferInputStream(ByteBufferUtils.combine(bufs))))
+ {
+ _messageBody = is.readObject();
+ }
+ catch (IOException e)
+ {
+ throw new ConnectionScopedRuntimeException("Unexpected IO Exception in operation in memory", e);
+ }
+ catch (ClassNotFoundException e)
+ {
+ throw new ConnectionScopedRuntimeException("Object message contained an object which could not " +
+ "be deserialized", e);
+ }
}
- catch (IOException e)
- {
- throw new ConnectionScopedRuntimeException("Unexpected IO Exception in operation in memory", e);
- }
- catch (ClassNotFoundException e)
+ else
{
- throw new ConnectionScopedRuntimeException("Object message contained an object which could not " +
- "be deserialized", e);
+ _messageBody = null;
}
_destinationName = destinationName;
}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/04b19e96/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/MessageConverter_Internal_to_v1_0.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/MessageConverter_Internal_to_v1_0.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/MessageConverter_Internal_to_v1_0.java
index 2ac734a..4a9e3ae 100644
--- a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/MessageConverter_Internal_to_v1_0.java
+++ b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/MessageConverter_Internal_to_v1_0.java
@@ -27,14 +27,17 @@ import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.List;
import java.util.Map;
+import java.util.UUID;
import org.apache.qpid.server.message.internal.InternalMessage;
import org.apache.qpid.server.plugin.PluggableService;
+import org.apache.qpid.server.protocol.converter.MessageConversionException;
import org.apache.qpid.server.protocol.v1_0.messaging.SectionEncoder;
import org.apache.qpid.server.protocol.v1_0.type.Binary;
import org.apache.qpid.server.protocol.v1_0.type.Symbol;
import org.apache.qpid.server.protocol.v1_0.type.UnsignedByte;
import org.apache.qpid.server.protocol.v1_0.type.UnsignedInteger;
+import org.apache.qpid.server.protocol.v1_0.type.UnsignedLong;
import org.apache.qpid.server.protocol.v1_0.type.messaging.AmqpValue;
import org.apache.qpid.server.protocol.v1_0.type.messaging.ApplicationProperties;
import org.apache.qpid.server.protocol.v1_0.type.messaging.Data;
@@ -71,9 +74,13 @@ public class MessageConverter_Internal_to_v1_0 extends MessageConverter_to_1_0<I
}
Properties properties = new Properties();
- properties.setCorrelationId(serverMessage.getMessageHeader().getCorrelationId());
+ if (serverMessage.getMessageHeader().getEncoding() != null)
+ {
+ properties.setContentEncoding(Symbol.valueOf(serverMessage.getMessageHeader().getEncoding()));
+ }
+ properties.setCorrelationId(getCorrelationId(serverMessage));
properties.setCreationTime(new Date(serverMessage.getMessageHeader().getTimestamp()));
- properties.setMessageId(serverMessage.getMessageHeader().getMessageId());
+ properties.setMessageId(getMessageId(serverMessage));
if(bodySection instanceof Data)
{
properties.setContentType(Symbol.valueOf(serverMessage.getMessageHeader().getMimeType()));
@@ -88,7 +95,15 @@ public class MessageConverter_Internal_to_v1_0 extends MessageConverter_to_1_0<I
ApplicationProperties applicationProperties = null;
if(!serverMessage.getMessageHeader().getHeaderNames().isEmpty())
{
- applicationProperties = new ApplicationProperties(serverMessage.getMessageHeader().getHeaderMap() );
+ try
+ {
+ applicationProperties = new ApplicationProperties(serverMessage.getMessageHeader().getHeaderMap());
+ }
+ catch (IllegalArgumentException e)
+ {
+ throw new MessageConversionException("Could not convert message from internal to 1.0"
+ + " because conversion of 'application headers' failed.", e);
+ }
}
return new MessageMetaData_1_0(header.createEncodingRetainingSection(),
@@ -102,6 +117,42 @@ public class MessageConverter_Internal_to_v1_0 extends MessageConverter_to_1_0<I
}
+ private Object getMessageId(final InternalMessage serverMessage)
+ {
+ String messageIdAsString = serverMessage.getMessageHeader().getMessageId();
+ return stringToMessageId(messageIdAsString);
+ }
+
+ private Object getCorrelationId(final InternalMessage serverMessage)
+ {
+ String correlationIdAsString = serverMessage.getMessageHeader().getCorrelationId();
+ return stringToMessageId(correlationIdAsString);
+ }
+
+ private Object stringToMessageId(final String correlationIdAsString)
+ {
+ Object messageId = null;
+ if (correlationIdAsString != null)
+ {
+ try
+ {
+ messageId = UUID.fromString(correlationIdAsString);
+ }
+ catch (IllegalArgumentException e)
+ {
+ try
+ {
+ messageId = UnsignedLong.valueOf(correlationIdAsString);
+ }
+ catch (NumberFormatException nfe)
+ {
+ messageId = correlationIdAsString;
+ }
+ }
+ }
+ return messageId;
+ }
+
@Override
protected EncodingRetainingSection<?> getBodySection(final InternalMessage serverMessage,
final SectionEncoder encoder)
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/04b19e96/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/MessageConverter_from_1_0.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/MessageConverter_from_1_0.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/MessageConverter_from_1_0.java
index cf35014..1cc4a22 100644
--- a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/MessageConverter_from_1_0.java
+++ b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/MessageConverter_from_1_0.java
@@ -317,6 +317,36 @@ public class MessageConverter_from_1_0
return userId;
}
+ public static String getReplyTo(final Message_1_0 serverMsg)
+ {
+ String replyTo = null;
+ final PropertiesSection propertiesSection = serverMsg.getPropertiesSection();
+ if (propertiesSection != null)
+ {
+ final Properties properties = propertiesSection.getValue();
+ if (properties != null)
+ {
+ replyTo = properties.getReplyTo();
+ }
+ }
+ return replyTo;
+ }
+
+ public static Symbol getContentEncoding(final Message_1_0 serverMsg)
+ {
+ Symbol contentEncoding = null;
+ final PropertiesSection propertiesSection = serverMsg.getPropertiesSection();
+ if (propertiesSection != null)
+ {
+ final Properties properties = propertiesSection.getValue();
+ if (properties != null)
+ {
+ contentEncoding = properties.getContentEncoding();
+ }
+ }
+ return contentEncoding;
+ }
+
public static Object getCorrelationId(final Message_1_0 serverMsg)
{
Object correlationIdObject = null;
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/04b19e96/broker-plugins/amqp-1-0-protocol/src/test/java/org/apache/qpid/server/protocol/v1_0/PropertyConverter_Internal_to_v1_0Test.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-1-0-protocol/src/test/java/org/apache/qpid/server/protocol/v1_0/PropertyConverter_Internal_to_v1_0Test.java b/broker-plugins/amqp-1-0-protocol/src/test/java/org/apache/qpid/server/protocol/v1_0/PropertyConverter_Internal_to_v1_0Test.java
new file mode 100644
index 0000000..455ab8f
--- /dev/null
+++ b/broker-plugins/amqp-1-0-protocol/src/test/java/org/apache/qpid/server/protocol/v1_0/PropertyConverter_Internal_to_v1_0Test.java
@@ -0,0 +1,307 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.server.protocol.v1_0;
+
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
+
+import org.apache.qpid.server.message.AMQMessageHeader;
+import org.apache.qpid.server.message.internal.InternalMessage;
+import org.apache.qpid.server.message.internal.InternalMessageHeader;
+import org.apache.qpid.server.message.internal.InternalMessageMetaData;
+import org.apache.qpid.server.message.internal.InternalMessageMetaDataType;
+import org.apache.qpid.server.model.NamedAddressSpace;
+import org.apache.qpid.server.protocol.converter.MessageConversionException;
+import org.apache.qpid.server.protocol.v1_0.type.Binary;
+import org.apache.qpid.server.protocol.v1_0.type.Symbol;
+import org.apache.qpid.server.protocol.v1_0.type.UnsignedLong;
+import org.apache.qpid.server.store.StoredMessage;
+import org.apache.qpid.test.utils.QpidTestCase;
+
+public class PropertyConverter_Internal_to_v1_0Test extends QpidTestCase
+{
+ private MessageConverter_Internal_to_v1_0 _messageConverter;
+ private NamedAddressSpace _addressSpace;
+
+ @Override
+ public void setUp() throws Exception
+ {
+ super.setUp();
+ _messageConverter = new MessageConverter_Internal_to_v1_0();
+ _addressSpace = mock(NamedAddressSpace.class);
+ }
+
+ public void testDurableTrueConversion()
+ {
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ InternalMessage originalMessage = createTestMessage(header, null, true);
+
+ Message_1_0 convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ assertTrue("Unexpected persistence of message", convertedMessage.isPersistent());
+ assertTrue("Unexpected persistence of meta data",
+ convertedMessage.getStoredMessage().getMetaData().isPersistent());
+ }
+
+ public void testDurableFalseConversion()
+ {
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ InternalMessage originalMessage = createTestMessage(header, null, false);
+
+ Message_1_0 convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ assertFalse("Unexpected persistence of message", convertedMessage.isPersistent());
+ assertFalse("Unexpected persistence of meta data",
+ convertedMessage.getStoredMessage().getMetaData().isPersistent());
+ }
+
+ public void testPriorityConversion()
+ {
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ byte priority = (byte) 7;
+ when(header.getPriority()).thenReturn(priority);
+ InternalMessage originalMessage = createTestMessage(header);
+
+ Message_1_0 convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ assertEquals("Unexpected priority", priority, convertedMessage.getMessageHeader().getPriority());
+ }
+
+ public void testExpirationConversion() throws InterruptedException
+ {
+ long ttl = 10000;
+ long expiryTime = System.currentTimeMillis() + ttl;
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ when(header.getExpiration()).thenReturn(expiryTime);
+ InternalMessage originalMessage = createTestMessage(header);
+ Thread.sleep(1L);
+
+ Message_1_0 convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ Long convertedTtl = MessageConverter_from_1_0.getTtl(convertedMessage);
+ assertEquals("Unexpected TTL", expiryTime - originalMessage.getArrivalTime(), convertedTtl.longValue());
+ }
+
+ public void testContentEncodingConversion()
+ {
+ String contentEncoding = "my-test-encoding";
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ when(header.getEncoding()).thenReturn(contentEncoding);
+ InternalMessage originalMessage = createTestMessage(header);
+
+ Message_1_0 convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ Symbol convertedContentEncoding = MessageConverter_from_1_0.getContentEncoding(convertedMessage);
+ assertEquals("Unexpected content encoding", contentEncoding, convertedContentEncoding.toString());
+ }
+
+ public void testMessageIdStringConversion()
+ {
+ final String messageId = "testMessageId";
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ when(header.getMessageId()).thenReturn(messageId);
+ InternalMessage originalMessage = createTestMessage(header);
+
+ Message_1_0 convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ Object convertedMessageId = MessageConverter_from_1_0.getMessageId(convertedMessage);
+ assertEquals("Unexpected messageId", messageId, convertedMessageId);
+ }
+
+ public void testMessageIdUuidConversion()
+ {
+ final UUID messageId = UUID.randomUUID();
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ when(header.getMessageId()).thenReturn(messageId.toString());
+ InternalMessage originalMessage = createTestMessage(header);
+
+ Message_1_0 convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ Object convertedMessageId = MessageConverter_from_1_0.getMessageId(convertedMessage);
+ assertEquals("Unexpected messageId", messageId, convertedMessageId);
+ }
+
+ public void testMessageIdUnsignedLongConversion()
+ {
+ final UnsignedLong messageId = UnsignedLong.valueOf(-1L);
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ when(header.getMessageId()).thenReturn(messageId.toString());
+ InternalMessage originalMessage = createTestMessage(header);
+
+ Message_1_0 convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ Object convertedMessageId = MessageConverter_from_1_0.getMessageId(convertedMessage);
+ assertEquals("Unexpected messageId", messageId, convertedMessageId);
+ }
+
+ public void testCorrelationIdStringConversion()
+ {
+ final String correlationId = "testCorrelationId";
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ when(header.getCorrelationId()).thenReturn(correlationId);
+ InternalMessage originalMessage = createTestMessage(header);
+
+ Message_1_0 convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ Object convertedCorrelationId = MessageConverter_from_1_0.getCorrelationId(convertedMessage);
+ assertEquals("Unexpected messageId", correlationId, convertedCorrelationId);
+ }
+
+ public void testCorrelationIdUuidConversion()
+ {
+ final UUID correlationId = UUID.randomUUID();
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ when(header.getCorrelationId()).thenReturn(correlationId.toString());
+ InternalMessage originalMessage = createTestMessage(header);
+
+ Message_1_0 convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ Object convertedCorrelationId = MessageConverter_from_1_0.getCorrelationId(convertedMessage);
+ assertEquals("Unexpected correlationId", correlationId, convertedCorrelationId);
+ }
+
+ public void testCorrelationIdUnsignedLongConversion()
+ {
+ final UnsignedLong correlationId = UnsignedLong.valueOf(-1L);
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ when(header.getCorrelationId()).thenReturn(correlationId.toString());
+ InternalMessage originalMessage = createTestMessage(header);
+
+ Message_1_0 convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ Object convertedCorrelationId = MessageConverter_from_1_0.getCorrelationId(convertedMessage);
+ assertEquals("Unexpected correlationId", correlationId, convertedCorrelationId);
+ }
+
+ public void testUserIdConversion()
+ {
+ final String userId = "testUserId";
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ when(header.getUserId()).thenReturn(userId);
+ InternalMessage originalMessage = createTestMessage(header);
+
+ Message_1_0 convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ Binary convertedUserId = MessageConverter_from_1_0.getUserId(convertedMessage);
+ assertTrue("Unexpected userId", Arrays.equals(userId.getBytes(UTF_8), convertedUserId.getArray()));
+ }
+
+ public void testReplyToConversion()
+ {
+ final String replyTo = "amq.direct/test";
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ when(header.getReplyTo()).thenReturn(replyTo);
+ InternalMessage originalMessage = createTestMessage(header);
+
+ Message_1_0 convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ String convertedReplyTo = MessageConverter_from_1_0.getReplyTo(convertedMessage);
+ assertEquals("Unexpected replyTo", replyTo, convertedReplyTo);
+ }
+
+ public void testTimestampConversion()
+ {
+ final long timestamp = System.currentTimeMillis();
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ when(header.getTimestamp()).thenReturn(timestamp);
+ InternalMessage originalMessage = createTestMessage(header);
+
+ Message_1_0 convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ Date creationTime = MessageConverter_from_1_0.getCreationTime(convertedMessage);
+ assertNotNull("timestamp not converted", creationTime);
+ assertEquals("Unexpected timestamp", timestamp, creationTime.getTime());
+ }
+
+ public void testHeadersConversion()
+ {
+ final Map<String, Object> properties = new HashMap<>();
+ properties.put("testProperty1", "testProperty1Value");
+ properties.put("intProperty", 1);
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ when(header.getHeaderNames()).thenReturn(properties.keySet());
+ doAnswer(invocation ->
+ {
+ final String originalArgument = (String) (invocation.getArguments())[0];
+ return properties.get(originalArgument);
+ }).when(header).getHeader(any(String.class));
+ InternalMessage originalMessage = createTestMessage(header);
+
+ final Message_1_0 convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ Map<String, Object> convertedHeaders = convertedMessage.getApplicationPropertiesSection().getValue();
+ assertEquals("Unexpected application properties", properties, new HashMap<>(convertedHeaders));
+ }
+
+ public void testHeadersConversionWithNonSimpleTypes()
+ {
+ final Map<String, Object> properties = Collections.singletonMap("listProperty", Collections.emptyList());
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ when(header.getHeaderNames()).thenReturn(properties.keySet());
+ doAnswer(invocation ->
+ {
+ final String originalArgument = (String) (invocation.getArguments())[0];
+ return properties.get(originalArgument);
+ }).when(header).getHeader(any(String.class));
+ InternalMessage originalMessage = createTestMessage(header);
+
+ try
+ {
+ _messageConverter.convert(originalMessage, _addressSpace);
+ fail("Expected exception not thrown");
+ }
+ catch (MessageConversionException e)
+ {
+ // pass
+ }
+ }
+
+ private InternalMessage createTestMessage(final AMQMessageHeader header)
+ {
+ return createTestMessage(header, null, false);
+ }
+
+ private InternalMessage createTestMessage(final AMQMessageHeader header,
+ byte[] content,
+ final boolean persistent)
+ {
+ final InternalMessageHeader internalMessageHeader = new InternalMessageHeader(header);
+ final int contentSize = content == null ? 0 : content.length;
+ final InternalMessageMetaData metaData =
+ new InternalMessageMetaData(persistent, internalMessageHeader, contentSize);
+ final StoredMessage<InternalMessageMetaData> storedMessage = mock(StoredMessage.class);
+
+ when(storedMessage.getMetaData()).thenReturn(metaData);
+ when(storedMessage.getContentSize()).thenReturn(contentSize);
+ return ((InternalMessage) InternalMessageMetaDataType.INSTANCE.createMessage(storedMessage));
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[3/3] qpid-broker-j git commit: QPID-7434: Add unit tests for
property conversion of internal messages to AMQP 0-8 messages
Posted by or...@apache.org.
QPID-7434: Add unit tests for property conversion of internal messages to AMQP 0-8 messages
Project: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/commit/8ab67e0c
Tree: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/tree/8ab67e0c
Diff: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/diff/8ab67e0c
Branch: refs/heads/master
Commit: 8ab67e0c145b5a18df424c829e99b47b63d65726
Parents: 392187c
Author: Alex Rudyy <or...@apache.org>
Authored: Tue Aug 1 17:08:58 2017 +0100
Committer: Alex Rudyy <or...@apache.org>
Committed: Tue Aug 1 17:08:58 2017 +0100
----------------------------------------------------------------------
.../v0_8/MessageConverter_Internal_to_v0_8.java | 47 ++-
.../PropertyConverter_Internal_to_v0_8Test.java | 336 +++++++++++++++++++
2 files changed, 378 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/8ab67e0c/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_Internal_to_v0_8.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_Internal_to_v0_8.java b/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_Internal_to_v0_8.java
index 82bbb29..a83f4df 100644
--- a/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_Internal_to_v0_8.java
+++ b/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_Internal_to_v0_8.java
@@ -32,6 +32,7 @@ import org.apache.qpid.server.message.mimecontentconverter.ObjectToMimeContentCo
import org.apache.qpid.server.model.NamedAddressSpace;
import org.apache.qpid.server.plugin.MessageConverter;
import org.apache.qpid.server.plugin.PluggableService;
+import org.apache.qpid.server.protocol.converter.MessageConversionException;
import org.apache.qpid.server.protocol.v0_8.transport.BasicContentHeaderProperties;
import org.apache.qpid.server.protocol.v0_8.transport.ContentHeaderBody;
import org.apache.qpid.server.protocol.v0_8.transport.MessagePublishInfo;
@@ -166,15 +167,17 @@ public class MessageConverter_Internal_to_v0_8 implements MessageConverter<Inter
final BasicContentHeaderProperties props = new BasicContentHeaderProperties();
props.setAppId(serverMsg.getMessageHeader().getAppId());
props.setContentType(bodyMimeType);
- props.setCorrelationId(serverMsg.getMessageHeader().getCorrelationId());
+ props.setCorrelationId(convertToShortStringForProperty("correlation-id", serverMsg.getMessageHeader().getCorrelationId()));
props.setDeliveryMode(serverMsg.isPersistent() ? BasicContentHeaderProperties.PERSISTENT : BasicContentHeaderProperties.NON_PERSISTENT);
props.setExpiration(serverMsg.getExpiration());
- props.setMessageId(serverMsg.getMessageHeader().getMessageId());
+ props.setMessageId(convertToOptionalAMQPShortString(serverMsg.getMessageHeader().getMessageId()));
props.setPriority(serverMsg.getMessageHeader().getPriority());
- props.setReplyTo(serverMsg.getMessageHeader().getReplyTo());
+ props.setReplyTo(convertToOptionalAMQPShortString(serverMsg.getMessageHeader().getReplyTo()));
props.setTimestamp(serverMsg.getMessageHeader().getTimestamp());
- props.setUserId(serverMsg.getMessageHeader().getUserId());
+ props.setUserId(convertToOptionalAMQPShortString(serverMsg.getMessageHeader().getUserId()));
+
+ props.setEncoding(convertToShortStringForProperty("encoding", serverMsg.getMessageHeader().getEncoding()));
Map<String,Object> headerProps = new LinkedHashMap<String, Object>();
@@ -183,13 +186,47 @@ public class MessageConverter_Internal_to_v0_8 implements MessageConverter<Inter
headerProps.put(headerName, serverMsg.getMessageHeader().getHeader(headerName));
}
- props.setHeaders(FieldTable.convertToFieldTable(headerProps));
+ try
+ {
+ props.setHeaders(FieldTable.convertToFieldTable(headerProps));
+ }
+ catch (IllegalArgumentException | AMQPInvalidClassException e)
+ {
+ throw new MessageConversionException("Could not convert message from internal to 0-8 because headers conversion failed.", e);
+ }
final ContentHeaderBody chb = new ContentHeaderBody(props);
chb.setBodySize(size);
return new MessageMetaData(publishInfo, chb, serverMsg.getArrivalTime());
}
+ private AMQShortString convertToOptionalAMQPShortString(final String stringValue)
+ {
+ AMQShortString result;
+ try
+ {
+ result = AMQShortString.valueOf(stringValue);
+ }
+ catch (IllegalArgumentException e)
+ {
+ result = null;
+ }
+ return result;
+ }
+
+ private AMQShortString convertToShortStringForProperty(String propertyName, String s)
+ {
+ try
+ {
+ return AMQShortString.valueOf(s);
+ }
+ catch (IllegalArgumentException e)
+ {
+ throw new MessageConversionException(String.format(
+ "Could not convert message from internal to 0-8 because conversion of '%s' failed.", propertyName), e);
+ }
+ }
+
@Override
public String getType()
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/8ab67e0c/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/PropertyConverter_Internal_to_v0_8Test.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/PropertyConverter_Internal_to_v0_8Test.java b/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/PropertyConverter_Internal_to_v0_8Test.java
new file mode 100644
index 0000000..6915812
--- /dev/null
+++ b/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/PropertyConverter_Internal_to_v0_8Test.java
@@ -0,0 +1,336 @@
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.server.protocol.v0_8;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
+
+import org.apache.qpid.server.message.AMQMessageHeader;
+import org.apache.qpid.server.message.internal.InternalMessage;
+import org.apache.qpid.server.message.internal.InternalMessageHeader;
+import org.apache.qpid.server.message.internal.InternalMessageMetaData;
+import org.apache.qpid.server.message.internal.InternalMessageMetaDataType;
+import org.apache.qpid.server.model.NamedAddressSpace;
+import org.apache.qpid.server.protocol.converter.MessageConversionException;
+import org.apache.qpid.server.protocol.v0_8.transport.BasicContentHeaderProperties;
+import org.apache.qpid.server.store.StoredMessage;
+import org.apache.qpid.test.utils.QpidTestCase;
+
+public class PropertyConverter_Internal_to_v0_8Test extends QpidTestCase
+{
+ private MessageConverter_Internal_to_v0_8 _messageConverter;
+ private NamedAddressSpace _addressSpace;
+
+ @Override
+ public void setUp() throws Exception
+ {
+ super.setUp();
+ _messageConverter = new MessageConverter_Internal_to_v0_8();
+ _addressSpace = mock(NamedAddressSpace.class);
+ }
+
+ public void testDurableTrueConversion()
+ {
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ InternalMessage originalMessage = createTestMessage(header, null, true);
+
+ AMQMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ assertEquals("Unexpected delivery mode",
+ BasicContentHeaderProperties.PERSISTENT,
+ convertedMessage.getContentHeaderBody().getProperties().getDeliveryMode());
+ assertTrue("Unexpected persistence of message", convertedMessage.isPersistent());
+ assertTrue("Unexpected persistence of meta data",
+ convertedMessage.getStoredMessage().getMetaData().isPersistent());
+ }
+
+ public void testDurableFalseConversion()
+ {
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ InternalMessage originalMessage = createTestMessage(header, null, false);
+
+ AMQMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ assertEquals("Unexpected delivery mode",
+ BasicContentHeaderProperties.NON_PERSISTENT,
+ convertedMessage.getContentHeaderBody().getProperties().getDeliveryMode());
+ assertFalse("Unexpected persistence of message", convertedMessage.isPersistent());
+ assertFalse("Unexpected persistence of meta data",
+ convertedMessage.getStoredMessage().getMetaData().isPersistent());
+ }
+
+ public void testPriorityConversion()
+ {
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ byte priority = (byte) 7;
+ when(header.getPriority()).thenReturn(priority);
+ InternalMessage originalMessage = createTestMessage(header);
+
+ AMQMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ assertEquals("Unexpected priority",
+ priority,
+ convertedMessage.getContentHeaderBody().getProperties().getPriority());
+ }
+
+ public void testExpirationConversion() throws InterruptedException
+ {
+ long ttl = 10000;
+ long expiryTime = System.currentTimeMillis() + ttl;
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ when(header.getExpiration()).thenReturn(expiryTime);
+ InternalMessage originalMessage = createTestMessage(header);
+
+ AMQMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ assertEquals("Unexpected expiration time",
+ expiryTime,
+ convertedMessage.getContentHeaderBody().getProperties().getExpiration());
+ }
+
+ public void testContentEncodingConversion()
+ {
+ String contentEncoding = "my-test-encoding";
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ when(header.getEncoding()).thenReturn(contentEncoding);
+ InternalMessage originalMessage = createTestMessage(header);
+
+ AMQMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ assertEquals("Unexpected content encoding",
+ contentEncoding,
+ convertedMessage.getContentHeaderBody().getProperties().getEncodingAsString());
+ }
+
+ public void testLongContentEncodingConversion()
+ {
+ String contentEncoding = generateLongString();
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ when(header.getEncoding()).thenReturn(contentEncoding);
+ InternalMessage originalMessage = createTestMessage(header);
+
+ try
+ {
+ _messageConverter.convert(originalMessage, _addressSpace);
+ fail("Expected exception is not thrown");
+ }
+ catch (MessageConversionException e)
+ {
+ // pass
+ }
+ }
+
+ public void testMessageIdConversion()
+ {
+ final String messageId = "testMessageId";
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ when(header.getMessageId()).thenReturn(messageId);
+ InternalMessage originalMessage = createTestMessage(header);
+
+ AMQMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ assertEquals("Unexpected messageId",
+ messageId,
+ convertedMessage.getContentHeaderBody().getProperties().getMessageIdAsString());
+ }
+
+ public void testMessageIdConversionWhenLengthExceeds255()
+ {
+ final String messageId = generateLongString();
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ when(header.getMessageId()).thenReturn(messageId);
+ InternalMessage originalMessage = createTestMessage(header);
+
+ AMQMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ assertNull("Unexpected messageId", convertedMessage.getContentHeaderBody().getProperties().getMessageId());
+ }
+
+ public void testCorrelationIdConversionWhenLengthExceeds255()
+ {
+ final String correlationId = generateLongString();
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ when(header.getCorrelationId()).thenReturn(correlationId);
+ InternalMessage originalMessage = createTestMessage(header);
+
+ try
+ {
+ _messageConverter.convert(originalMessage, _addressSpace);
+ fail("Expected exception is not thrown");
+ }
+ catch (MessageConversionException e)
+ {
+ // pass
+ }
+ }
+
+ public void testUserIdConversion()
+ {
+ final String userId = "testUserId";
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ when(header.getUserId()).thenReturn(userId);
+ InternalMessage originalMessage = createTestMessage(header);
+
+ AMQMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ assertEquals("Unexpected userId",
+ userId,
+ convertedMessage.getContentHeaderBody().getProperties().getUserIdAsString());
+ }
+
+ public void testUserIdConversionWhenLengthExceeds255()
+ {
+ final String userId = generateLongString();
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ when(header.getUserId()).thenReturn(userId);
+ InternalMessage originalMessage = createTestMessage(header);
+
+ AMQMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ assertNull("Unexpected userId", convertedMessage.getContentHeaderBody().getProperties().getUserId());
+ }
+
+ public void testTimestampConversion()
+ {
+ final long timestamp = System.currentTimeMillis();
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ when(header.getTimestamp()).thenReturn(timestamp);
+ InternalMessage originalMessage = createTestMessage(header);
+
+ AMQMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ assertEquals("Unexpected timestamp",
+ timestamp,
+ convertedMessage.getContentHeaderBody().getProperties().getTimestamp());
+ }
+
+ public void testHeadersConversion()
+ {
+ final Map<String, Object> properties = new HashMap<>();
+ properties.put("testProperty1", "testProperty1Value");
+ properties.put("intProperty", 1);
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ when(header.getHeaderNames()).thenReturn(properties.keySet());
+ doAnswer(invocation ->
+ {
+ final String originalArgument = (String) (invocation.getArguments())[0];
+ return properties.get(originalArgument);
+ }).when(header).getHeader(any(String.class));
+ InternalMessage originalMessage = createTestMessage(header);
+
+ final AMQMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ Map<String, Object> convertedHeaders =
+ FieldTable.convertToMap(convertedMessage.getContentHeaderBody().getProperties().getHeaders());
+ assertEquals("Unexpected application properties", properties, new HashMap<>(convertedHeaders));
+ }
+
+ public void testHeadersConversionWithUnsupportedTypes()
+ {
+ final Map<String, Object> properties = Collections.singletonMap("uuidProperty", UUID.randomUUID());
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ when(header.getHeaderNames()).thenReturn(properties.keySet());
+ doAnswer(invocation ->
+ {
+ final String originalArgument = (String) (invocation.getArguments())[0];
+ return properties.get(originalArgument);
+ }).when(header).getHeader(any(String.class));
+ InternalMessage originalMessage = createTestMessage(header);
+
+ try
+ {
+ _messageConverter.convert(originalMessage, _addressSpace);
+ fail("Expected exception not thrown");
+ }
+ catch (MessageConversionException e)
+ {
+ // pass
+ }
+ }
+
+
+ public void testHeadersConversionWhenKeyLengthExceeds255()
+ {
+ final Map<String, Object> properties = Collections.singletonMap(generateLongString(), "test");
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ when(header.getHeaderNames()).thenReturn(properties.keySet());
+ doAnswer(invocation ->
+ {
+ final String originalArgument = (String) (invocation.getArguments())[0];
+ return properties.get(originalArgument);
+ }).when(header).getHeader(any(String.class));
+ InternalMessage originalMessage = createTestMessage(header);
+
+ try
+ {
+ _messageConverter.convert(originalMessage, _addressSpace);
+ fail("Expected exception not thrown");
+ }
+ catch (MessageConversionException e)
+ {
+ // pass
+ }
+ }
+
+ private InternalMessage createTestMessage(final AMQMessageHeader header)
+ {
+ return createTestMessage(header, null, false);
+ }
+
+ private InternalMessage createTestMessage(final AMQMessageHeader header,
+ byte[] content,
+ final boolean persistent)
+ {
+ final InternalMessageHeader internalMessageHeader = new InternalMessageHeader(header);
+ final int contentSize = content == null ? 0 : content.length;
+ final InternalMessageMetaData metaData =
+ new InternalMessageMetaData(persistent, internalMessageHeader, contentSize);
+ final StoredMessage<InternalMessageMetaData> storedMessage = mock(StoredMessage.class);
+
+ when(storedMessage.getMetaData()).thenReturn(metaData);
+ when(storedMessage.getContentSize()).thenReturn(contentSize);
+ return ((InternalMessage) InternalMessageMetaDataType.INSTANCE.createMessage(storedMessage));
+ }
+
+ private String generateLongString()
+ {
+ return generateLongString(AMQShortString.MAX_LENGTH + 1);
+ }
+
+ private String generateLongString(int stringLength)
+ {
+ StringBuilder buffer = new StringBuilder();
+ for (int i = 0; i < stringLength; i++)
+ {
+ buffer.append('x');
+ }
+
+ return buffer.toString();
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[2/3] qpid-broker-j git commit: QPID-7434: Add unit tests for
property conversion of AMQP 0-8..0-9-1 messages to internal messages
Posted by or...@apache.org.
QPID-7434: Add unit tests for property conversion of AMQP 0-8..0-9-1 messages to internal messages
Project: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/commit/392187c1
Tree: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/tree/392187c1
Diff: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/diff/392187c1
Branch: refs/heads/master
Commit: 392187c183c5b4c8e897aced80ef5e3933be39f5
Parents: 04b19e9
Author: Alex Rudyy <or...@apache.org>
Authored: Tue Aug 1 16:13:46 2017 +0100
Committer: Alex Rudyy <or...@apache.org>
Committed: Tue Aug 1 16:13:46 2017 +0100
----------------------------------------------------------------------
.../v0_8/MessageConverter_v0_8_to_Internal.java | 6 +-
.../MessageConverter_v0_8_to_InternalTest.java | 322 +++++++++++++++++++
2 files changed, 325 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/392187c1/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_v0_8_to_Internal.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_v0_8_to_Internal.java b/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_v0_8_to_Internal.java
index d16a239..19a937d 100644
--- a/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_v0_8_to_Internal.java
+++ b/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_v0_8_to_Internal.java
@@ -244,19 +244,19 @@ public class MessageConverter_v0_8_to_Internal implements MessageConverter<AMQMe
AMQBindingURL burl = new AMQBindingURL(origReplyToString);
ReplyToComponents replyTo = new ReplyToComponents();
String routingKey = burl.getRoutingKey();
- if(routingKey != null)
+ if(routingKey != null && !"".equals(routingKey))
{
replyTo.setRoutingKey(routingKey);
}
String exchangeName = burl.getExchangeName();
- if(exchangeName != null)
+ if(exchangeName != null && !"".equals(exchangeName))
{
replyTo.setExchange(exchangeName);
}
String queueName = burl.getQueueName();
- if(queueName != null)
+ if(queueName != null && !"".equals(queueName))
{
replyTo.setQueue(queueName);
}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/392187c1/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_v0_8_to_InternalTest.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_v0_8_to_InternalTest.java b/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_v0_8_to_InternalTest.java
new file mode 100644
index 0000000..cd22fd6
--- /dev/null
+++ b/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_v0_8_to_InternalTest.java
@@ -0,0 +1,322 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.qpid.server.protocol.v0_8;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.server.message.internal.InternalMessage;
+import org.apache.qpid.server.model.NamedAddressSpace;
+import org.apache.qpid.server.protocol.v0_8.transport.BasicContentHeaderProperties;
+import org.apache.qpid.server.protocol.v0_8.transport.ContentHeaderBody;
+import org.apache.qpid.server.protocol.v0_8.transport.MessagePublishInfo;
+import org.apache.qpid.server.store.StoredMessage;
+import org.apache.qpid.test.utils.QpidTestCase;
+
+public class MessageConverter_v0_8_to_InternalTest extends QpidTestCase
+{
+ private MessageConverter_v0_8_to_Internal _messageConverter;
+ private NamedAddressSpace _addressSpace;
+
+ @Override
+ public void setUp() throws Exception
+ {
+ super.setUp();
+ _messageConverter = new MessageConverter_v0_8_to_Internal();
+ _addressSpace = mock(NamedAddressSpace.class);
+ }
+
+ public void testDeliveryModePersistentConversion()
+ {
+ BasicContentHeaderProperties header = new BasicContentHeaderProperties();
+ header.setDeliveryMode(BasicContentHeaderProperties.PERSISTENT);
+ final AMQMessage originalMessage = createTestMessage(header);
+
+ InternalMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ assertTrue("Unexpected persistence of message", convertedMessage.isPersistent());
+ assertTrue("Unexpected persistence of meta data",
+ convertedMessage.getStoredMessage().getMetaData().isPersistent());
+ }
+
+ public void testDeliveryModeNonPersistentConversion()
+ {
+ BasicContentHeaderProperties header = new BasicContentHeaderProperties();
+ header.setDeliveryMode(BasicContentHeaderProperties.NON_PERSISTENT);
+ final AMQMessage originalMessage = createTestMessage(header);
+
+ InternalMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ assertFalse("Unexpected persistence of message", convertedMessage.isPersistent());
+ assertFalse("Unexpected persistence of meta data",
+ convertedMessage.getStoredMessage().getMetaData().isPersistent());
+ }
+
+ public void testPriorityConversion()
+ {
+ byte priority = (byte) 7;
+ BasicContentHeaderProperties header = new BasicContentHeaderProperties();
+ header.setPriority(priority);
+ final AMQMessage originalMessage = createTestMessage(header);
+
+ InternalMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ assertEquals("Unexpected priority", priority, convertedMessage.getMessageHeader().getPriority());
+ }
+
+ public void testExpirationConversion()
+ {
+ long ttl = 10000;
+ long arrivalTime = System.currentTimeMillis();
+ long expiryTime = arrivalTime + ttl;
+ BasicContentHeaderProperties header = new BasicContentHeaderProperties();
+ header.setExpiration(expiryTime);
+ final AMQMessage originalMessage = createTestMessage(header, arrivalTime);
+
+ InternalMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ assertEquals("Unexpected expiration", expiryTime, convertedMessage.getMessageHeader().getExpiration());
+ }
+
+ public void testContentEncodingConversion()
+ {
+ String contentEncoding = "my-test-encoding";
+ BasicContentHeaderProperties header = new BasicContentHeaderProperties();
+ header.setEncoding(contentEncoding);
+ final AMQMessage originalMessage = createTestMessage(header);
+
+ InternalMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ assertEquals("Unexpected content encoding", contentEncoding, convertedMessage.getMessageHeader().getEncoding());
+ }
+
+ public void testMessageIdConversion()
+ {
+ final String messageId = "testMessageId";
+ BasicContentHeaderProperties header = new BasicContentHeaderProperties();
+ header.setMessageId(messageId);
+ final AMQMessage originalMessage = createTestMessage(header);
+
+ InternalMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ assertEquals("Unexpected messageId",
+ messageId,
+ convertedMessage.getMessageHeader().getMessageId());
+ }
+
+ public void testCorrelationIdStringConversion()
+ {
+ final String correlationId = "testMessageCorrelationId";
+ BasicContentHeaderProperties header = new BasicContentHeaderProperties();
+ header.setCorrelationId(correlationId);
+ final AMQMessage originalMessage = createTestMessage(header);
+
+ InternalMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ assertEquals("Unexpected correlationId",
+ correlationId,
+ convertedMessage.getMessageHeader().getCorrelationId());
+ }
+
+ public void testUserIdConversion()
+ {
+ final String userId = "testUserId";
+ BasicContentHeaderProperties header = new BasicContentHeaderProperties();
+ header.setUserId(userId);
+ final AMQMessage originalMessage = createTestMessage(header);
+
+ InternalMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ assertEquals("Unexpected userId", userId, convertedMessage.getMessageHeader().getUserId());
+ }
+
+ public void testReplyToConversionForDirectExchangeAndRoutingKey()
+ {
+ String exchangeName = "amq.direct";
+ String routingKey = "testRoutingKey";
+ final String replyTo = String.format("%s://%s//?routingkey='%s'", "direct", exchangeName, routingKey);
+ BasicContentHeaderProperties header = new BasicContentHeaderProperties();
+ header.setReplyTo(replyTo);
+ final AMQMessage originalMessage = createTestMessage(header);
+
+ InternalMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ assertEquals("Unexpected replyTo",
+ exchangeName + "/" + routingKey,
+ convertedMessage.getMessageHeader().getReplyTo());
+ }
+
+ public void testReplyToConversionForFanoutExchange()
+ {
+ String exchangeName = "amq.fanout";
+ final String replyTo = String.format("%s://%s//", "fanout", exchangeName);
+ BasicContentHeaderProperties header = new BasicContentHeaderProperties();
+ header.setReplyTo(replyTo);
+ final AMQMessage originalMessage = createTestMessage(header);
+
+ InternalMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ assertEquals("Unexpected replyTo", exchangeName, convertedMessage.getMessageHeader().getReplyTo());
+ }
+
+ public void testReplyToConversionForDefaultDestination()
+ {
+ String exchangeName = "";
+ String routingKey = "testRoutingKey";
+ final String replyTo = String.format("%s://%s//?routingkey='%s'", "direct", exchangeName, routingKey);
+ BasicContentHeaderProperties header = new BasicContentHeaderProperties();
+ header.setReplyTo(replyTo);
+ final AMQMessage originalMessage = createTestMessage(header);
+
+ InternalMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ assertEquals("Unexpected replyTo", routingKey, convertedMessage.getMessageHeader().getReplyTo());
+ }
+
+ public void testReplyToNonBurl()
+ {
+ final String replyTo = "test/routing";
+ BasicContentHeaderProperties header = new BasicContentHeaderProperties();
+ header.setReplyTo(replyTo);
+ final AMQMessage originalMessage = createTestMessage(header);
+
+ InternalMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ assertEquals("Unexpected replyTo", replyTo, convertedMessage.getMessageHeader().getReplyTo());
+ }
+
+ public void testTimestampConversion()
+ {
+ final long creationTime = System.currentTimeMillis();
+ BasicContentHeaderProperties header = new BasicContentHeaderProperties();
+ header.setTimestamp(creationTime);
+ final AMQMessage originalMessage = createTestMessage(header);
+
+ InternalMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ assertEquals("Unexpected timestamp", creationTime, convertedMessage.getMessageHeader().getTimestamp());
+ }
+
+ public void testHeadersConversion()
+ {
+ Map<String, Object> properties = new HashMap<>();
+ properties.put("testProperty1", "testProperty1Value");
+ properties.put("intProperty", 1);
+ BasicContentHeaderProperties header = new BasicContentHeaderProperties();
+ header.setHeaders(FieldTable.convertToFieldTable(properties));
+ final AMQMessage originalMessage = createTestMessage(header);
+
+ final InternalMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ final Map<String, Object> headers = convertedMessage.getMessageHeader().getHeaderMap();
+ assertEquals("Unexpected headers", properties, new HashMap<>(headers));
+ }
+
+ public void testContentTypeConversion()
+ {
+ final String contentType = "text/json";
+ BasicContentHeaderProperties header = new BasicContentHeaderProperties();
+ header.setContentType(contentType);
+ final AMQMessage originalMessage = createTestMessage(header);
+
+ InternalMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ assertEquals("Unexpected content type", contentType, convertedMessage.getMessageHeader().getMimeType());
+ }
+
+ public void testTypeConversion()
+ {
+ final String type = "JMSType";
+ BasicContentHeaderProperties header = new BasicContentHeaderProperties();
+ header.setType(type);
+ final AMQMessage originalMessage = createTestMessage(header);
+
+ InternalMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ assertEquals("Unexpected type", type, convertedMessage.getMessageHeader().getType());
+ }
+
+ public void testApplicationIdConversion()
+ {
+ final String applicationId = "appId";
+ BasicContentHeaderProperties header = new BasicContentHeaderProperties();
+ header.setAppId(applicationId);
+ final AMQMessage originalMessage = createTestMessage(header);
+
+ InternalMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ assertEquals("Unexpected applicationId", applicationId, convertedMessage.getMessageHeader().getAppId());
+ }
+
+ public void testBasicPublishConversion()
+ {
+ final String exchangeName = "amq.direct";
+ final String testRoutingKey = "test-routing-key";
+
+ final AMQMessage originalMessage = createTestMessage(new BasicContentHeaderProperties());
+ originalMessage.getMessagePublishInfo().setRoutingKey(AMQShortString.valueOf(testRoutingKey));
+ originalMessage.getMessagePublishInfo().setExchange(AMQShortString.valueOf(exchangeName));
+
+ InternalMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ assertEquals("Unexpected to", exchangeName, convertedMessage.getTo());
+
+ // TODO: QPID-7868 : add test for initialRoutingAddress
+ }
+
+ private AMQMessage createTestMessage(final BasicContentHeaderProperties basicContentHeaderProperties)
+ {
+ return createTestMessage(basicContentHeaderProperties, null, 0);
+ }
+
+ private AMQMessage createTestMessage(final BasicContentHeaderProperties basicContentHeaderProperties,
+ long arrivalTime)
+ {
+ return createTestMessage(basicContentHeaderProperties, null, arrivalTime);
+ }
+
+ private AMQMessage createTestMessage(final BasicContentHeaderProperties basicContentHeaderProperties,
+ final byte[] content, final long arrivalTime)
+ {
+ final ContentHeaderBody contentHeaderBody = mock(ContentHeaderBody.class);
+ when(contentHeaderBody.getProperties()).thenReturn(basicContentHeaderProperties);
+
+ final StoredMessage<MessageMetaData> storedMessage = mock(StoredMessage.class);
+ when(storedMessage.getMetaData()).thenReturn(new MessageMetaData(new MessagePublishInfo(),
+ contentHeaderBody,
+ arrivalTime));
+
+ if (content != null)
+ {
+ when(storedMessage.getContentSize()).thenReturn(content.length);
+ when(storedMessage.getContent(0, content.length)).thenReturn(Collections.singleton(QpidByteBuffer.wrap(
+ content)));
+ }
+
+ return new AMQMessage(storedMessage);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org