Repository: qpid-broker-j Updated Branches: refs/heads/master 42bebb9ff -> 17e6c7d6e
QPID-7434: [Java Broker] Improve Internal to AMQP 0-x message conversion and add unit tests 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/17e6c7d6 Tree: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/tree/17e6c7d6 Diff: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/diff/17e6c7d6 Branch: refs/heads/master Commit: 17e6c7d6e070ecfa8a46edb643607b0c6ef9d099 Parents: 939cda5 Author: Lorenz Quack <lqu...@apache.org> Authored: Fri Aug 11 14:30:37 2017 +0100 Committer: Lorenz Quack <lqu...@apache.org> Committed: Fri Aug 11 15:14:53 2017 +0100 ---------------------------------------------------------------------- .../protocol/v0_10/transport/EncoderUtils.java | 2 +- .../ListToAmqpListConverter.java | 3 +- .../MessageConverter_Internal_to_0_10Test.java | 278 ++++++++++++++++++ .../v0_8/MessageConverter_Internal_to_v0_8.java | 11 +- .../MessageConverter_Internal_to_0_8Test.java | 280 +++++++++++++++++++ 5 files changed, 571 insertions(+), 3 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/17e6c7d6/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/transport/EncoderUtils.java ---------------------------------------------------------------------- diff --git a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/transport/EncoderUtils.java b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/transport/EncoderUtils.java index 32c5e6d..63b2c09 100644 --- a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/transport/EncoderUtils.java +++ b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/transport/EncoderUtils.java @@ -357,7 +357,7 @@ public class EncoderUtils if (key instanceof String) { String string = (String)key; - if ( string.length() > 0xFF) + if (string.length() > 0xFF) { return false; } http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/17e6c7d6/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/transport/mimecontentconverter/ListToAmqpListConverter.java ---------------------------------------------------------------------- diff --git a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/transport/mimecontentconverter/ListToAmqpListConverter.java b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/transport/mimecontentconverter/ListToAmqpListConverter.java index 3dfc06b..17717b1 100644 --- a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/transport/mimecontentconverter/ListToAmqpListConverter.java +++ b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/transport/mimecontentconverter/ListToAmqpListConverter.java @@ -26,6 +26,7 @@ import java.util.List; import org.apache.qpid.server.message.mimecontentconverter.ObjectToMimeContentConverter; import org.apache.qpid.server.plugin.PluggableService; import org.apache.qpid.server.protocol.v0_10.transport.BBEncoder; +import org.apache.qpid.server.protocol.v0_10.transport.EncoderUtils; @PluggableService public class ListToAmqpListConverter implements ObjectToMimeContentConverter<List> @@ -57,7 +58,7 @@ public class ListToAmqpListConverter implements ObjectToMimeContentConverter<Lis @Override public boolean isAcceptable(final List list) { - return true; + return EncoderUtils.isEncodable(list); } @Override http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/17e6c7d6/broker-plugins/amqp-0-10-protocol/src/test/java/org/apache/qpid/server/protocol/v0_10/MessageConverter_Internal_to_0_10Test.java ---------------------------------------------------------------------- diff --git a/broker-plugins/amqp-0-10-protocol/src/test/java/org/apache/qpid/server/protocol/v0_10/MessageConverter_Internal_to_0_10Test.java b/broker-plugins/amqp-0-10-protocol/src/test/java/org/apache/qpid/server/protocol/v0_10/MessageConverter_Internal_to_0_10Test.java new file mode 100644 index 0000000..aeb9a79 --- /dev/null +++ b/broker-plugins/amqp-0-10-protocol/src/test/java/org/apache/qpid/server/protocol/v0_10/MessageConverter_Internal_to_0_10Test.java @@ -0,0 +1,278 @@ +/* + * 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_10; + +import static java.nio.charset.StandardCharsets.UTF_8; +import static org.junit.Assert.assertArrayEquals; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.io.ByteArrayOutputStream; +import java.io.ObjectOutputStream; +import java.io.Serializable; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; + +import com.google.common.collect.Lists; +import com.google.common.io.ByteStreams; +import org.mockito.ArgumentCaptor; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.stubbing.Answer; + +import org.apache.qpid.server.bytebuffer.QpidByteBuffer; +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.v0_10.transport.mimecontentconverter.ListToAmqpListConverter; +import org.apache.qpid.server.protocol.v0_10.transport.mimecontentconverter.MapToAmqpMapConverter; +import org.apache.qpid.server.store.StoredMessage; +import org.apache.qpid.server.typedmessage.mimecontentconverter.ListToJmsStreamMessage; +import org.apache.qpid.server.typedmessage.mimecontentconverter.MapToJmsMapMessage; +import org.apache.qpid.test.utils.QpidTestCase; + +public class MessageConverter_Internal_to_0_10Test extends QpidTestCase +{ + private final MessageConverter_Internal_to_v0_10 _converter = new MessageConverter_Internal_to_v0_10(); + private final StoredMessage<InternalMessageMetaData> _handle = mock(StoredMessage.class); + private final AMQMessageHeader _amqpHeader = mock(AMQMessageHeader.class); + + @Override + public void setUp() throws Exception + { + super.setUp(); + } + + + public void testStringMessage() throws Exception + { + String content = "testContent"; + final String mimeType = "text/plain"; + doTest(content, mimeType, content.getBytes(UTF_8), mimeType); + } + + public void testStringMessageWithUnknownMimeType() throws Exception + { + String content = "testContent"; + final String mimeType = "foo/bar"; + doTest(content, mimeType, content.getBytes(UTF_8), "text/plain"); + } + + public void testStringMessageWithoutMimeType() throws Exception + { + String content = "testContent"; + doTest(content, null, content.getBytes(UTF_8), "text/plain"); + } + + public void testListMessageWithMimeType() throws Exception + { + ArrayList<?> content = Lists.newArrayList("testItem", 37.5, 42); + final ListToJmsStreamMessage listToJmsStreamMessage = new ListToJmsStreamMessage(); + final byte[] expectedContent = listToJmsStreamMessage.toMimeContent(content); + doTest(content, "foo/bar", expectedContent, listToJmsStreamMessage.getMimeType()); + } + + public void testListMessageWithoutMimeType() throws Exception + { + ArrayList<?> content = Lists.newArrayList("testItem", 37.5, 42); + final ListToJmsStreamMessage listToJmsStreamMessage = new ListToJmsStreamMessage(); + final byte[] expectedContent = listToJmsStreamMessage.toMimeContent(content); + doTest(content, null, expectedContent, listToJmsStreamMessage.getMimeType()); + } + + public void testListMessageWithoutMimeTypeWithNonJmsContent() throws Exception + { + ArrayList<?> content = Lists.newArrayList("testItem", 37.5, 42, Lists.newArrayList()); + final ListToAmqpListConverter listToAmqpListConverter = new ListToAmqpListConverter(); + final byte[] expectedContent = listToAmqpListConverter.toMimeContent(content); + doTest(content, null, expectedContent, listToAmqpListConverter.getMimeType()); + } + + public void testListMessageWithoutMimeTypeWithNonConvertibleItem() throws Exception + { + ArrayList<?> content = Lists.newArrayList(new MySerializable()); + final InternalMessage sourceMessage = getAmqMessage(content, null); + doTest(content, null, getObjectStreamMessageBytes(content), "application/java-object-stream"); + } + + public void testByteArrayMessageWithoutMimeType() throws Exception + { + byte[] content = "testContent".getBytes(UTF_8); + doTest(content, null, content, "application/octet-stream"); + } + + public void testByteArrayMessageWithMimeType() throws Exception + { + byte[] content = "testContent".getBytes(UTF_8); + final String mimeType = "foo/bar"; + doTest(content, mimeType, content, mimeType); + } + + public void testEmptyByteArrayMessageWithMimeType() throws Exception + { + byte[] content = new byte[0]; + final String mimeType = "foo/bar"; + doTest(content, mimeType, content, mimeType); + } + + public void testMapMessageWithMimeType() throws Exception + { + HashMap<Object, Object> content = new HashMap<>(); + content.put("key1", 37); + content.put("key2", "foo"); + final String mimeType = "foo/bar"; + final MapToJmsMapMessage mapToJmsMapMessage = new MapToJmsMapMessage(); + final byte[] expectedContent = mapToJmsMapMessage.toMimeContent(content); + doTest(content, mimeType, expectedContent, mapToJmsMapMessage.getMimeType()); + } + + public void testMapMessageWithoutMimeType() throws Exception + { + HashMap<Object, Object> content = new HashMap<>(); + content.put("key1", 37); + content.put("key2", "foo"); + final MapToJmsMapMessage mapToJmsMapMessage = new MapToJmsMapMessage(); + final byte[] expectedContent = mapToJmsMapMessage.toMimeContent(content); + doTest(content, null, expectedContent, mapToJmsMapMessage.getMimeType()); + } + + public void testMapMessageWithMimeTypeWithNonJmsContent() throws Exception + { + HashMap<Object, Object> content = new HashMap<>(); + content.put("key", Collections.singletonMap("foo", "bar")); + final String mimeType = "foo/bar"; + final MapToAmqpMapConverter mapToAmqpMapConverter = new MapToAmqpMapConverter(); + final byte[] expectedContent = mapToAmqpMapConverter.toMimeContent(content); + doTest(content, mimeType, expectedContent, mapToAmqpMapConverter.getMimeType()); + } + + public void testMapMessageWithoutMimeTypeWithNonConvertibleEntry() throws Exception + { + HashMap<Object, Object> content = new HashMap<>(); + content.put(37, new MySerializable()); + + doTest(content, null, getObjectStreamMessageBytes(content), "application/java-object-stream"); + } + + public void testSerializableMessageWithMimeType() throws Exception + { + Serializable content = new MySerializable(); + final String mimeType = "foo/bar"; + doTest(content, mimeType, getObjectStreamMessageBytes(content), "application/java-object-stream"); + } + + public void testSerializableMessageWithoutMimeType() throws Exception + { + Serializable content = new MySerializable(); + doTest(content, null, getObjectStreamMessageBytes(content), "application/java-object-stream"); + } + + public void testNullMessageWithoutMimeType() throws Exception + { + doTest(null, null, null, null); + } + + + private byte[] getObjectStreamMessageBytes(final Serializable o) throws Exception + { + try (ByteArrayOutputStream bos = new ByteArrayOutputStream(); + ObjectOutputStream oos = new ObjectOutputStream(bos)) + { + oos.writeObject(o); + return bos.toByteArray(); + } + } + + + protected InternalMessage getAmqMessage(final Serializable content, final String mimeType) throws Exception + { + final byte[] serializedContent = getObjectStreamMessageBytes(content); + configureMessageContent(serializedContent); + configureMessageHeader(mimeType); + + final InternalMessageHeader internalMessageHeader = new InternalMessageHeader(_amqpHeader); + final int contentSize = serializedContent == null ? 0 : serializedContent.length; + final InternalMessageMetaData metaData = + new InternalMessageMetaData(false, internalMessageHeader, contentSize); + when(_handle.getMetaData()).thenReturn(metaData); + + return ((InternalMessage) InternalMessageMetaDataType.INSTANCE.createMessage(_handle)); + } + + private void configureMessageHeader(final String mimeType) + { + when(_amqpHeader.getMimeType()).thenReturn(mimeType); + } + + private void configureMessageContent(byte[] section) + { + if (section == null) + { + section = new byte[0]; + } + final QpidByteBuffer combined = QpidByteBuffer.wrap(section); + when(_handle.getContentSize()).thenReturn(section.length); + final ArgumentCaptor<Integer> offsetCaptor = ArgumentCaptor.forClass(Integer.class); + final ArgumentCaptor<Integer> sizeCaptor = ArgumentCaptor.forClass(Integer.class); + + when(_handle.getContent(offsetCaptor.capture(), + sizeCaptor.capture())).then(new Answer<Collection<QpidByteBuffer>>() + { + @Override + public Collection<QpidByteBuffer> answer(final InvocationOnMock invocation) throws Throwable + { + final QpidByteBuffer view = combined.view(offsetCaptor.getValue(), sizeCaptor.getValue()); + return Collections.singleton(view); + } + }); + } + + private void doTest(final Serializable messageBytes, + final String mimeType, + final byte[] expectedContent, + final String expectedContentType) throws Exception + { + final InternalMessage sourceMessage = getAmqMessage(messageBytes, mimeType); + final MessageTransferMessage convertedMessage = _converter.convert(sourceMessage, mock(NamedAddressSpace.class)); + final Collection<QpidByteBuffer> content = convertedMessage.getContent(0, (int) convertedMessage.getSize()); + + assertArrayEquals("Unexpected content", expectedContent != null ? expectedContent : new byte[0], getBytes(content)); + assertEquals("Unexpected content type", expectedContentType, convertedMessage.getMessageHeader().getMimeType()); + } + + private byte[] getBytes(final Collection<QpidByteBuffer> content) throws Exception + { + ByteArrayOutputStream bos = new ByteArrayOutputStream(); + for (QpidByteBuffer buf : content) + { + ByteStreams.copy(buf.asInputStream(), bos); + buf.dispose(); + } + return bos.toByteArray(); + } + + private static class MySerializable implements Serializable + { + } +} http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/17e6c7d6/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 a83f4df..a8ff769 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 @@ -71,7 +71,16 @@ public class MessageConverter_Internal_to_v0_8 implements MessageConverter<Inter { Object messageBody = serverMsg.getMessageBody(); ObjectToMimeContentConverter converter = MimeContentConverterRegistry.getBestFitObjectToMimeContentConverter(messageBody); - final byte[] messageContent = converter == null ? new byte[] {} : converter.toMimeContent(messageBody); + final byte[] messageContent; + try + { + messageContent = converter == null ? new byte[] {} : converter.toMimeContent(messageBody); + } + catch (IllegalArgumentException e) + { + throw new MessageConversionException("Could not convert message from Internal to 0-8 because" + + " conversion of message content failed.", e); + } String mimeType = converter == null ? null : converter.getMimeType(); mimeType = improveMimeType(serverMsg, mimeType); http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/17e6c7d6/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_Internal_to_0_8Test.java ---------------------------------------------------------------------- diff --git a/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_Internal_to_0_8Test.java b/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_Internal_to_0_8Test.java new file mode 100644 index 0000000..6de6ceb --- /dev/null +++ b/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_Internal_to_0_8Test.java @@ -0,0 +1,280 @@ +/* + * 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 java.nio.charset.StandardCharsets.UTF_8; +import static org.junit.Assert.assertArrayEquals; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.io.ByteArrayOutputStream; +import java.io.ObjectOutputStream; +import java.io.Serializable; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.UUID; + +import com.google.common.collect.Lists; +import com.google.common.io.ByteStreams; +import org.mockito.ArgumentCaptor; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.stubbing.Answer; + +import org.apache.qpid.server.bytebuffer.QpidByteBuffer; +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_10.transport.mimecontentconverter.ListToAmqpListConverter; +import org.apache.qpid.server.protocol.v0_10.transport.mimecontentconverter.MapToAmqpMapConverter; +import org.apache.qpid.server.store.StoredMessage; +import org.apache.qpid.server.typedmessage.mimecontentconverter.ListToJmsStreamMessage; +import org.apache.qpid.server.typedmessage.mimecontentconverter.MapToJmsMapMessage; +import org.apache.qpid.test.utils.QpidTestCase; + +public class MessageConverter_Internal_to_0_8Test extends QpidTestCase +{ + private final MessageConverter_Internal_to_v0_8 _converter = new MessageConverter_Internal_to_v0_8(); + private final StoredMessage<InternalMessageMetaData> _handle = mock(StoredMessage.class); + private final AMQMessageHeader _amqpHeader = mock(AMQMessageHeader.class); + + @Override + public void setUp() throws Exception + { + super.setUp(); + } + + + public void testStringMessage() throws Exception + { + String content = "testContent"; + final String mimeType = "text/plain"; + doTest(content, mimeType, content.getBytes(UTF_8), mimeType); + } + + public void testStringMessageWithUnknownMimeType() throws Exception + { + String content = "testContent"; + final String mimeType = "foo/bar"; + doTest(content, mimeType, content.getBytes(UTF_8), "text/plain"); + } + + public void testStringMessageWithoutMimeType() throws Exception + { + String content = "testContent"; + doTest(content, null, content.getBytes(UTF_8), "text/plain"); + } + + public void testListMessageWithMimeType() throws Exception + { + ArrayList<?> content = Lists.newArrayList("testItem", 37.5, 42); + final ListToJmsStreamMessage listToJmsStreamMessage = new ListToJmsStreamMessage(); + final byte[] expectedContent = listToJmsStreamMessage.toMimeContent(content); + doTest(content, "foo/bar", expectedContent, listToJmsStreamMessage.getMimeType()); + } + + public void testListMessageWithoutMimeType() throws Exception + { + ArrayList<?> content = Lists.newArrayList("testItem", 37.5, 42); + final ListToJmsStreamMessage listToJmsStreamMessage = new ListToJmsStreamMessage(); + final byte[] expectedContent = listToJmsStreamMessage.toMimeContent(content); + doTest(content, null, expectedContent, listToJmsStreamMessage.getMimeType()); + } + + public void testListMessageWithoutMimeTypeWithNonJmsContent() throws Exception + { + ArrayList<?> content = Lists.newArrayList("testItem", 37.5, 42, Lists.newArrayList()); + final ListToAmqpListConverter listToAmqpListConverter = new ListToAmqpListConverter(); + final byte[] expectedContent = listToAmqpListConverter.toMimeContent(content); + doTest(content, null, expectedContent, listToAmqpListConverter.getMimeType()); + } + + public void testListMessageWithoutMimeTypeWithNonConvertibleItem() throws Exception + { + ArrayList<?> content = Lists.newArrayList(new MySerializable()); + final InternalMessage sourceMessage = getAmqMessage(content, null); + doTest(content, null, getObjectStreamMessageBytes(content), "application/java-object-stream"); + } + + public void testByteArrayMessageWithoutMimeType() throws Exception + { + byte[] content = "testContent".getBytes(UTF_8); + doTest(content, null, content, "application/octet-stream"); + } + + public void testByteArrayMessageWithMimeType() throws Exception + { + byte[] content = "testContent".getBytes(UTF_8); + final String mimeType = "foo/bar"; + doTest(content, mimeType, content, mimeType); + } + + public void testEmptyByteArrayMessageWithMimeType() throws Exception + { + byte[] content = new byte[0]; + final String mimeType = "foo/bar"; + doTest(content, mimeType, content, mimeType); + } + + public void testMapMessageWithMimeType() throws Exception + { + HashMap<Object, Object> content = new HashMap<>(); + content.put("key1", 37); + content.put("key2", "foo"); + final String mimeType = "foo/bar"; + final MapToJmsMapMessage mapToJmsMapMessage = new MapToJmsMapMessage(); + final byte[] expectedContent = mapToJmsMapMessage.toMimeContent(content); + doTest(content, mimeType, expectedContent, mapToJmsMapMessage.getMimeType()); + } + + public void testMapMessageWithoutMimeType() throws Exception + { + HashMap<Object, Object> content = new HashMap<>(); + content.put("key1", 37); + content.put("key2", "foo"); + final MapToJmsMapMessage mapToJmsMapMessage = new MapToJmsMapMessage(); + final byte[] expectedContent = mapToJmsMapMessage.toMimeContent(content); + doTest(content, null, expectedContent, mapToJmsMapMessage.getMimeType()); + } + + public void testMapMessageWithMimeTypeWithNonJmsContent() throws Exception + { + HashMap<Object, Object> content = new HashMap<>(); + content.put("key", Collections.singletonMap("foo", "bar")); + final String mimeType = "foo/bar"; + final MapToAmqpMapConverter mapToAmqpMapConverter = new MapToAmqpMapConverter(); + final byte[] expectedContent = mapToAmqpMapConverter.toMimeContent(content); + doTest(content, mimeType, expectedContent, mapToAmqpMapConverter.getMimeType()); + } + + public void testMapMessageWithoutMimeTypeWithNonConvertibleEntry() throws Exception + { + HashMap<Object, Object> content = new HashMap<>(); + content.put(37, new MySerializable()); + + doTest(content, null, getObjectStreamMessageBytes(content), "application/java-object-stream"); + } + + public void testSerializableMessageWithMimeType() throws Exception + { + Serializable content = new MySerializable(); + final String mimeType = "foo/bar"; + doTest(content, mimeType, getObjectStreamMessageBytes(content), "application/java-object-stream"); + } + + public void testSerializableMessageWithoutMimeType() throws Exception + { + Serializable content = new MySerializable(); + doTest(content, null, getObjectStreamMessageBytes(content), "application/java-object-stream"); + } + + public void testNullMessageWithoutMimeType() throws Exception + { + doTest(null, null, null, null); + } + + + private byte[] getObjectStreamMessageBytes(final Serializable o) throws Exception + { + try (ByteArrayOutputStream bos = new ByteArrayOutputStream(); + ObjectOutputStream oos = new ObjectOutputStream(bos)) + { + oos.writeObject(o); + return bos.toByteArray(); + } + } + + + protected InternalMessage getAmqMessage(final Serializable content, final String mimeType) throws Exception + { + final byte[] serializedContent = getObjectStreamMessageBytes(content); + configureMessageContent(serializedContent); + configureMessageHeader(mimeType); + + final InternalMessageHeader internalMessageHeader = new InternalMessageHeader(_amqpHeader); + final int contentSize = serializedContent == null ? 0 : serializedContent.length; + final InternalMessageMetaData metaData = + new InternalMessageMetaData(false, internalMessageHeader, contentSize); + when(_handle.getMetaData()).thenReturn(metaData); + + return ((InternalMessage) InternalMessageMetaDataType.INSTANCE.createMessage(_handle)); + } + + private void configureMessageHeader(final String mimeType) + { + when(_amqpHeader.getMimeType()).thenReturn(mimeType); + } + + private void configureMessageContent(byte[] section) + { + if (section == null) + { + section = new byte[0]; + } + final QpidByteBuffer combined = QpidByteBuffer.wrap(section); + when(_handle.getContentSize()).thenReturn(section.length); + final ArgumentCaptor<Integer> offsetCaptor = ArgumentCaptor.forClass(Integer.class); + final ArgumentCaptor<Integer> sizeCaptor = ArgumentCaptor.forClass(Integer.class); + + when(_handle.getContent(offsetCaptor.capture(), + sizeCaptor.capture())).then(new Answer<Collection<QpidByteBuffer>>() + { + @Override + public Collection<QpidByteBuffer> answer(final InvocationOnMock invocation) throws Throwable + { + final QpidByteBuffer view = combined.view(offsetCaptor.getValue(), sizeCaptor.getValue()); + return Collections.singleton(view); + } + }); + } + + private void doTest(final Serializable messageBytes, + final String mimeType, + final byte[] expectedContent, + final String expectedContentType) throws Exception + { + final InternalMessage sourceMessage = getAmqMessage(messageBytes, mimeType); + final AMQMessage convertedMessage = _converter.convert(sourceMessage, mock(NamedAddressSpace.class)); + final Collection<QpidByteBuffer> content = convertedMessage.getContent(0, (int) convertedMessage.getSize()); + + assertArrayEquals("Unexpected content", expectedContent != null ? expectedContent : new byte[0], getBytes(content)); + assertEquals("Unexpected content type", expectedContentType, convertedMessage.getMessageHeader().getMimeType()); + } + + private byte[] getBytes(final Collection<QpidByteBuffer> content) throws Exception + { + ByteArrayOutputStream bos = new ByteArrayOutputStream(); + for (QpidByteBuffer buf : content) + { + ByteStreams.copy(buf.asInputStream(), bos); + buf.dispose(); + } + return bos.toByteArray(); + } + + private static class MySerializable implements Serializable + { + } +} --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@qpid.apache.org For additional commands, e-mail: commits-h...@qpid.apache.org
