From 73d9c6b6c5392b83ec70b73cfbe0bc0ecd53e7bf Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Sun, 21 May 2017 12:56:41 -0400 Subject: [PATCH] INT-4255: JSON Embedded Headers Message Mapper JIRA: https://jira.spring.io/browse/INT-4255 - Support embedding headers for transports that don't support headers (TCP, Kafka, etc) - Use the new message-aware Jackson ObjectMapper - Provide a mechanism to more efficiently support byte[] payloads (avoid Base64 encoding) - Support decoding "legacy" SCSt embedded headers Polishing; PR Comments - Add Support to MQTT and TCP Switch to simple patterns instead of regex * Fix JavaDoc typo * Upgrade to Jackson 2.9.1 --- build.gradle | 2 +- .../mapping/BytesMessageMapper.java | 29 +++ .../EmbeddedJsonHeadersMessageMapper.java | 244 ++++++++++++++++++ .../json/JsonInboundMessageMapper.java | 5 +- .../json/JsonOutboundMessageMapper.java | 10 +- ...EmbeddedJsonHeadersMessageMapperTests.java | 117 +++++++++ .../ip/tcp/connection/TcpMessageMapper.java | 32 ++- .../tcp/connection/TcpMessageMapperTests.java | 27 +- .../support/DefaultPahoMessageConverter.java | 66 ++++- .../mqtt/BackToBackAdapterTests.java | 90 +++++++ 10 files changed, 602 insertions(+), 20 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/mapping/BytesMessageMapper.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/support/json/EmbeddedJsonHeadersMessageMapper.java create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/support/json/EmbeddedJsonHeadersMessageMapperTests.java diff --git a/build.gradle b/build.gradle index aeb95fb5b7..932c77e692 100644 --- a/build.gradle +++ b/build.gradle @@ -105,7 +105,7 @@ subprojects { subproject -> hibernateVersion = '5.2.10.Final' hsqldbVersion = '2.4.0' h2Version = '1.4.196' - jackson2Version = '2.9.0' + jackson2Version = '2.9.1' javaxActivationVersion = '1.1.1' javaxMailVersion = '1.6.0' jedisVersion = '2.9.0' diff --git a/spring-integration-core/src/main/java/org/springframework/integration/mapping/BytesMessageMapper.java b/spring-integration-core/src/main/java/org/springframework/integration/mapping/BytesMessageMapper.java new file mode 100644 index 0000000000..88a6f6604b --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/mapping/BytesMessageMapper.java @@ -0,0 +1,29 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed 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.springframework.integration.mapping; + +/** + * An {@link OutboundMessageMapper} and {@link InboundMessageMapper} that + * maps to/from {@code byte[]}. + * + * @author Gary Russell + * @since 5.0 + * + */ +public interface BytesMessageMapper extends InboundMessageMapper, OutboundMessageMapper { + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/json/EmbeddedJsonHeadersMessageMapper.java b/spring-integration-core/src/main/java/org/springframework/integration/support/json/EmbeddedJsonHeadersMessageMapper.java new file mode 100644 index 0000000000..abb2a73cbb --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/json/EmbeddedJsonHeadersMessageMapper.java @@ -0,0 +1,244 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed 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.springframework.integration.support.json; + +import java.nio.ByteBuffer; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.integration.mapping.BytesMessageMapper; +import org.springframework.integration.support.MutableMessage; +import org.springframework.integration.support.MutableMessageHeaders; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.util.PatternMatchUtils; + +import com.fasterxml.jackson.databind.ObjectMapper; + +/** + * For outbound messages, uses a message-aware Jackson object mapper to render the message + * as JSON. For messages with {@code byte[]} payloads, if rendered as JSON, Jackson + * performs Base64 conversion on the bytes. If the {@link #setRawBytes(boolean) rawBytes} + * property is true (default), the result has the form + * <headersLen><headers><payloadLen><payload>; with the headers + * rendered in JSON and the payload unchanged. + *

+ * By default, all headers are included; you can provide simple patterns to specify a + * subset of headers. + *

+ * If neither expected format is detected, or an error occurs during conversion, the + * payload of the message is the original {@code byte[]}. + *

+ * IMPORTANT + *

+ * The default object mapper will only deserialize classes in certain packages. + * + * 

+ *	"java.util",
+ *	"java.lang",
+ *	"org.springframework.messaging.support",
+ *	"org.springframework.integration.support",
+ *	"org.springframework.integration.message",
+ *	"org.springframework.integration.store"
+ *
+ *

+ * To add more packages, create an object mapper using + * {@link JacksonJsonUtils#messagingAwareMapper(String...)}. + *

+ * A constructor is provided allowing the provision of such a configured object mapper. + * + * @author Gary Russell + * + * @since 5.0 + * + */ +public class EmbeddedJsonHeadersMessageMapper implements BytesMessageMapper { + + protected final Log logger = LogFactory.getLog(getClass()); + + private final ObjectMapper objectMapper; + + private final List headerPatterns; + + private final boolean allHeaders; + + private boolean rawBytes = true; + + private boolean caseSensitive; + + /** + * Construct an instance that embeds all headers, using the default + * JSON Object mapper. + */ + public EmbeddedJsonHeadersMessageMapper() { + this("*"); + } + + /** + * Construct an instance that embeds headers matching the supplied patterns, using + * the default JSON object mapper. + * @param headerPatterns the patterns. + * @see PatternMatchUtils#simpleMatch(String, String) + */ + public EmbeddedJsonHeadersMessageMapper(String... headerPatterns) { + this(JacksonJsonUtils.messagingAwareMapper(), headerPatterns); + } + + /** + * Construct an instance that embeds all headers, using the + * supplied JSON object mapper. + * @param objectMapper the object mapper. + */ + public EmbeddedJsonHeadersMessageMapper(ObjectMapper objectMapper) { + this(objectMapper, "*"); + } + + /** + * Construct an instance that embeds headers matching the supplied patterns using the + * supplied JSON object mapper. + * @param objectMapper the object mapper. + * @param headerPatterns the patterns. + */ + public EmbeddedJsonHeadersMessageMapper(ObjectMapper objectMapper, String... headerPatterns) { + this.objectMapper = objectMapper; + this.headerPatterns = Arrays.asList(headerPatterns); + this.allHeaders = this.headerPatterns.size() == 1 && this.headerPatterns.get(0).equals("*"); + } + + /** + * For messages with {@code byte[]} payloads, if rendered as JSON, Jackson performs + * Base64 conversion on the bytes. If this property is true (default), the result has + * the form <headersLen><headers><payloadLen><payload>; with + * the headers rendered in JSON and the payload unchanged. Set to false to render + * the bytes as base64. + * @param rawBytes false to encode as base64. + */ + public void setRawBytes(boolean rawBytes) { + this.rawBytes = rawBytes; + } + + /** + * Set to true to make the header name pattern match case sensitive. + * Default false. + * @param caseSensitive true to make case sensitive. + */ + public void setCaseSensitive(boolean caseSensitive) { + this.caseSensitive = caseSensitive; + } + + public Collection getHeaderPatterns() { + return Collections.unmodifiableCollection(this.headerPatterns); + } + + @SuppressWarnings("unchecked") + @Override + public byte[] fromMessage(Message message) throws Exception { + Message messageToEncode = this.allHeaders ? message : pruneHeaders(message); + if (this.rawBytes && message.getPayload() instanceof byte[]) { + return fromBytesPayload((Message) messageToEncode); + } + else { + return this.objectMapper.writeValueAsBytes(messageToEncode); + } + } + + private Message pruneHeaders(Message message) { + Map headersToEmbed = + message.getHeaders().entrySet().stream() + .filter(e -> this.headerPatterns.stream().anyMatch(p -> + this.caseSensitive + ? PatternMatchUtils.simpleMatch(p, e.getKey()) + : PatternMatchUtils.simpleMatch(p.toLowerCase(), e.getKey().toLowerCase()))) + .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); + return new MutableMessage<>(message.getPayload(), headersToEmbed); + } + + private byte[] fromBytesPayload(Message message) throws Exception { + byte[] headers = this.objectMapper.writeValueAsBytes(message.getHeaders()); + byte[] payload = message.getPayload(); + ByteBuffer buffer = ByteBuffer.wrap(new byte[8 + headers.length + payload.length]); + buffer.putInt(headers.length); + buffer.put(headers); + buffer.putInt(payload.length); + buffer.put(payload); + return buffer.array(); + } + + @Override + public Message toMessage(byte[] bytes) throws Exception { + Message message = null; + try { + message = decodeNativeFormat(bytes); + } + catch (Exception e) { + // empty + } + if (message == null) { + try { + message = (Message) this.objectMapper.readValue(bytes, Object.class); + } + catch (Exception e) { + if (this.logger.isDebugEnabled()) { + this.logger.debug("Failed to decode JSON", e); + } + } + } + if (message != null) { + return message; + } + else { + return new GenericMessage<>(bytes); + } + } + + private Message decodeNativeFormat(byte[] bytes) throws Exception { + ByteBuffer buffer = ByteBuffer.wrap(bytes); + if (buffer.remaining() > 4) { + int headersLen = buffer.getInt(); + if (headersLen >= 0 && headersLen < buffer.remaining() - 4) { + buffer.position(headersLen + 4); + int payloadLen = buffer.getInt(); + if (payloadLen != buffer.remaining()) { + return null; + } + else { + buffer.position(4); + @SuppressWarnings("unchecked") + Map headers = this.objectMapper.readValue(bytes, buffer.position(), headersLen, + HashMap.class); + buffer.position(buffer.position() + headersLen); + buffer.getInt(); + Object payload; + byte[] payloadBytes = new byte[payloadLen]; + buffer.get(payloadBytes); + payload = payloadBytes; + return new GenericMessage(payload, new MutableMessageHeaders(headers)); + } + } + } + return null; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/json/JsonInboundMessageMapper.java b/spring-integration-core/src/main/java/org/springframework/integration/support/json/JsonInboundMessageMapper.java index d03809566b..3ef52c5948 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/json/JsonInboundMessageMapper.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/json/JsonInboundMessageMapper.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -26,6 +26,8 @@ import org.springframework.util.Assert; /** * {@link org.springframework.integration.mapping.InboundMessageMapper} implementation that maps incoming JSON messages * to a {@link Message} with the specified payload type. + *

+ * Consider using the {@link EmbeddedJsonHeadersMessageMapper} instead. * * @author Jeremy Grelle * @author Oleg Zhurakousky @@ -60,6 +62,7 @@ public class JsonInboundMessageMapper extends AbstractJsonInboundMessageMapper toMessage(String jsonMessage) throws Exception { return this.messageParser.doInParser(this, jsonMessage); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/json/JsonOutboundMessageMapper.java b/spring-integration-core/src/main/java/org/springframework/integration/support/json/JsonOutboundMessageMapper.java index daa6806a73..676f47596a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/json/JsonOutboundMessageMapper.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/json/JsonOutboundMessageMapper.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -21,11 +21,16 @@ import org.springframework.messaging.Message; import org.springframework.util.Assert; /** - * {@link OutboundMessageMapper} implementation the converts a {@link Message} to a JSON string representation. + * {@link OutboundMessageMapper} implementation the converts a {@link Message} to a JSON + * string representation. + *

+ * Consider using the {@link EmbeddedJsonHeadersMessageMapper} instead; it provides more + * flexibility for determining which headers are included. * * @author Jeremy Grelle * @author Mark Fisher * @author Artem Bilan + * @author Gary Russell * @since 2.0 */ public class JsonOutboundMessageMapper implements OutboundMessageMapper { @@ -47,6 +52,7 @@ public class JsonOutboundMessageMapper implements OutboundMessageMapper this.shouldExtractPayload = shouldExtractPayload; } + @Override public String fromMessage(Message message) throws Exception { return this.jsonObjectMapper.toJson(this.shouldExtractPayload ? message.getPayload() : message); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/support/json/EmbeddedJsonHeadersMessageMapperTests.java b/spring-integration-core/src/test/java/org/springframework/integration/support/json/EmbeddedJsonHeadersMessageMapperTests.java new file mode 100644 index 0000000000..37ffb015b5 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/support/json/EmbeddedJsonHeadersMessageMapperTests.java @@ -0,0 +1,117 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed 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.springframework.integration.support.json; + +import static org.hamcrest.CoreMatchers.containsString; +import static org.hamcrest.CoreMatchers.equalTo; +import static org.hamcrest.CoreMatchers.not; +import static org.junit.Assert.assertThat; + +import java.nio.ByteBuffer; +import java.util.Collections; + +import org.junit.Test; + +import org.springframework.messaging.Message; +import org.springframework.messaging.support.GenericMessage; + +/** + * @author Gary Russell + * @since 5.0 + * + */ +public class EmbeddedJsonHeadersMessageMapperTests { + + @Test + public void testEmbedAll() throws Exception { + EmbeddedJsonHeadersMessageMapper mapper = new EmbeddedJsonHeadersMessageMapper(); + GenericMessage message = new GenericMessage<>("foo"); + assertThat(mapper.toMessage(mapper.fromMessage(message)), equalTo(message)); + + } + + @Test + public void testEmbedSome() throws Exception { + EmbeddedJsonHeadersMessageMapper mapper = new EmbeddedJsonHeadersMessageMapper("id"); + GenericMessage message = new GenericMessage<>("foo"); + Message decoded = mapper.toMessage(mapper.fromMessage(message)); + assertThat(decoded.getPayload(), equalTo(message.getPayload())); + assertThat(decoded.getHeaders().getId(), equalTo(message.getHeaders().getId())); + } + + @Test + public void testBytesEmbedAll() throws Exception { + EmbeddedJsonHeadersMessageMapper mapper = new EmbeddedJsonHeadersMessageMapper(); + GenericMessage message = new GenericMessage<>("foo".getBytes()); + Thread.sleep(2); + byte[] bytes = mapper.fromMessage(message); + ByteBuffer bb = ByteBuffer.wrap(bytes); + int headerLen = bb.getInt(); + byte[] headerBytes = new byte[headerLen]; + bb.get(headerBytes); + String headers = new String(headerBytes); + assertThat(headers, containsString(message.getHeaders().getId().toString())); + assertThat(headers, containsString(String.valueOf(message.getHeaders().getTimestamp()))); + assertThat(bb.getInt(), equalTo(3)); + assertThat(bb.remaining(), equalTo(3)); + assertThat((char) bb.get(), equalTo('f')); + assertThat((char) bb.get(), equalTo('o')); + assertThat((char) bb.get(), equalTo('o')); + } + + @Test + public void testBytesEmbedSome() throws Exception { + EmbeddedJsonHeadersMessageMapper mapper = new EmbeddedJsonHeadersMessageMapper("I*"); + GenericMessage message = new GenericMessage<>("foo".getBytes(), Collections.singletonMap("bar", "baz")); + byte[] bytes = mapper.fromMessage(message); + ByteBuffer bb = ByteBuffer.wrap(bytes); + int headerLen = bb.getInt(); + byte[] headerBytes = new byte[headerLen]; + bb.get(headerBytes); + String headers = new String(headerBytes); + assertThat(headers, containsString(message.getHeaders().getId().toString())); + assertThat(headers, not(containsString("bar"))); + assertThat(bb.getInt(), equalTo(3)); + assertThat(bb.remaining(), equalTo(3)); + assertThat((char) bb.get(), equalTo('f')); + assertThat((char) bb.get(), equalTo('o')); + assertThat((char) bb.get(), equalTo('o')); + } + + @Test + public void testBytesEmbedAllJson() throws Exception { + EmbeddedJsonHeadersMessageMapper mapper = new EmbeddedJsonHeadersMessageMapper(); + mapper.setRawBytes(false); + GenericMessage message = new GenericMessage<>("foo".getBytes()); + byte[] mappedBytes = mapper.fromMessage(message); + String mapped = new String(mappedBytes); + assertThat(mapped, containsString("[B\",\"Zm9v")); + @SuppressWarnings("unchecked") + Message decoded = (Message) mapper.toMessage(mappedBytes); + assertThat(new String(decoded.getPayload()), equalTo("foo")); + + } + + @Test + public void testBytesDecodeAll() throws Exception { + EmbeddedJsonHeadersMessageMapper mapper = new EmbeddedJsonHeadersMessageMapper(); + GenericMessage message = new GenericMessage<>("foo".getBytes()); + Message decoded = mapper.toMessage(mapper.fromMessage(message)); + assertThat(decoded, equalTo(message)); + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapper.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapper.java index 1319f11902..d542e2b8c3 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapper.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapper.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -26,6 +26,7 @@ import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.integration.ip.IpHeaders; +import org.springframework.integration.mapping.BytesMessageMapper; import org.springframework.integration.mapping.InboundMessageMapper; import org.springframework.integration.mapping.OutboundMessageMapper; import org.springframework.integration.support.AbstractIntegrationMessageBuilder; @@ -78,6 +79,8 @@ public class TcpMessageMapper implements private BeanFactory beanFactory; + private BytesMessageMapper bytesMessageMapper; + /** * Set the charset to use when converting outbound String messages to {@code byte[]}. * @param charset the charset to set @@ -89,7 +92,9 @@ public class TcpMessageMapper implements /** * Sets whether outbound String payloads are to be converted * to byte[]. Default is true. + * Ignored if a {@link BytesMessageMapper} is provided. * @param stringToBytes The stringToBytes to set. + * @see #setBytesMessageMapper(BytesMessageMapper) */ public void setStringToBytes(boolean stringToBytes) { this.stringToBytes = stringToBytes; @@ -138,6 +143,18 @@ public class TcpMessageMapper implements this.beanFactory = beanFactory; } + /** + * Set a {@link BytesMessageMapper} to use when mapping byte[]. + * {@link #setStringToBytes(boolean)} is ignored when a {@link BytesMessageMapper} + * is provided. + * @param bytesMessageMapper the mapper. + * @since 5.0 + * @see #setStringToBytes(boolean) + */ + public void setBytesMessageMapper(BytesMessageMapper bytesMessageMapper) { + this.bytesMessageMapper = bytesMessageMapper; + } + protected MessageBuilderFactory getMessageBuilderFactory() { if (!this.messageBuilderFactorySet) { if (this.beanFactory != null) { @@ -148,12 +165,20 @@ public class TcpMessageMapper implements return this.messageBuilderFactory; } + @SuppressWarnings("unchecked") @Override public Message toMessage(TcpConnection connection) throws Exception { Message message = null; Object payload = connection.getPayload(); if (payload != null) { - AbstractIntegrationMessageBuilder messageBuilder = getMessageBuilderFactory().withPayload(payload); + AbstractIntegrationMessageBuilder messageBuilder; + if (this.bytesMessageMapper != null && payload instanceof byte[]) { + messageBuilder = (AbstractIntegrationMessageBuilder) getMessageBuilderFactory() + .fromMessage(this.bytesMessageMapper.toMessage((byte[]) payload)); + } + else { + messageBuilder = getMessageBuilderFactory().withPayload(payload); + } this.addStandardHeaders(connection, messageBuilder); this.addCustomHeaders(connection, messageBuilder); message = messageBuilder.build(); @@ -208,6 +233,9 @@ public class TcpMessageMapper implements @Override public Object fromMessage(Message message) throws Exception { + if (this.bytesMessageMapper != null) { + return this.bytesMessageMapper.fromMessage(message); + } if (this.stringToBytes) { return getPayloadAsBytes(message); } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapperTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapperTests.java index 5a9d97380b..d6af49e62b 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapperTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapperTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -50,6 +50,7 @@ import org.springframework.integration.ip.IpHeaders; import org.springframework.integration.ip.tcp.serializer.MapJsonSerializer; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.support.converter.MapMessageConverter; +import org.springframework.integration.support.json.EmbeddedJsonHeadersMessageMapper; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.converter.MessageConverter; @@ -437,4 +438,28 @@ public class TcpMessageMapperTests { assertEquals("someId", message.getHeaders().get(IpHeaders.CONNECTION_ID)); } + @Test + public void testWithBytesMapper() throws Exception { + Message outMessage = MessageBuilder.withPayload("foo") + .setHeader("bar", "baz") + .build(); + TcpMessageMapper mapper = new TcpMessageMapper(); + mapper.setBytesMessageMapper(new EmbeddedJsonHeadersMessageMapper()); + byte[] bytes = (byte[]) mapper.fromMessage(outMessage); + + TcpConnection connection = mock(TcpConnection.class); + when(connection.getPayload()).thenReturn(bytes); + when(connection.getHostName()).thenReturn("someHost"); + when(connection.getHostAddress()).thenReturn("1.1.1.1"); + when(connection.getPort()).thenReturn(1234); + when(connection.getConnectionId()).thenReturn("someId"); + Message message = mapper.toMessage(connection); + assertEquals("foo", message.getPayload()); + assertEquals("baz", message.getHeaders().get("bar")); + assertEquals("someHost", message.getHeaders().get(IpHeaders.HOSTNAME)); + assertEquals("1.1.1.1", message.getHeaders().get(IpHeaders.IP_ADDRESS)); + assertEquals(1234, message.getHeaders().get(IpHeaders.REMOTE_PORT)); + assertEquals("someId", message.getHeaders().get(IpHeaders.CONNECTION_ID)); + } + } diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/DefaultPahoMessageConverter.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/DefaultPahoMessageConverter.java index 0ac7ab0aa1..d886724542 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/DefaultPahoMessageConverter.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/DefaultPahoMessageConverter.java @@ -21,11 +21,13 @@ import org.eclipse.paho.client.mqttv3.MqttMessage; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.integration.handler.MessageProcessor; +import org.springframework.integration.mapping.BytesMessageMapper; import org.springframework.integration.support.AbstractIntegrationMessageBuilder; import org.springframework.integration.support.DefaultMessageBuilderFactory; import org.springframework.integration.support.MessageBuilderFactory; import org.springframework.integration.support.utils.IntegrationUtils; import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.converter.MessageConversionException; import org.springframework.util.Assert; @@ -51,6 +53,8 @@ public class DefaultPahoMessageConverter implements MqttMessageConverter, BeanFa private final MessageProcessor retainedProcessor; + private BytesMessageMapper bytesMessageMapper; + private volatile boolean payloadAsBytes = false; private volatile BeanFactory beanFactory; @@ -164,8 +168,10 @@ public class DefaultPahoMessageConverter implements MqttMessageConverter, BeanFa /** * True if the converter should not convert the message payload to a String. + * Ignored if a {@link BytesMessageMapper} is provided. * * @param payloadAsBytes The payloadAsBytes to set. + * @see #setBytesMessageMapper(BytesMessageMapper) */ public void setPayloadAsBytes(boolean payloadAsBytes) { this.payloadAsBytes = payloadAsBytes; @@ -175,6 +181,18 @@ public class DefaultPahoMessageConverter implements MqttMessageConverter, BeanFa return this.payloadAsBytes; } + /** + * Set a {@link BytesMessageMapper} to use when mapping byte[]. + * {@link #setPayloadAsBytes(boolean)} is ignored when a {@link BytesMessageMapper} + * is provided. + * @param bytesMessageMapper the mapper. + * @since 5.0 + * @see #setPayloadAsBytes(boolean) + */ + public void setBytesMessageMapper(BytesMessageMapper bytesMessageMapper) { + this.bytesMessageMapper = bytesMessageMapper; + } + @Override public Message toMessage(Object mqttMessage, MessageHeaders headers) { Assert.isInstanceOf(MqttMessage.class, mqttMessage, @@ -183,11 +201,20 @@ public class DefaultPahoMessageConverter implements MqttMessageConverter, BeanFa return toMessage(null, (MqttMessage) mqttMessage); } + @SuppressWarnings("unchecked") @Override public Message toMessage(String topic, MqttMessage mqttMessage) { try { - AbstractIntegrationMessageBuilder messageBuilder = getMessageBuilderFactory() - .withPayload(mqttBytesToPayload(mqttMessage)) + AbstractIntegrationMessageBuilder messageBuilder; + if (this.bytesMessageMapper != null) { + messageBuilder = (AbstractIntegrationMessageBuilder) getMessageBuilderFactory() + .fromMessage(this.bytesMessageMapper.toMessage(mqttMessage.getPayload())); + } + else { + messageBuilder = getMessageBuilderFactory() + .withPayload(mqttBytesToPayload(mqttMessage)); + } + messageBuilder .setHeader(MqttHeaders.RECEIVED_QOS, mqttMessage.getQos()) .setHeader(MqttHeaders.DUPLICATE, mqttMessage.isDuplicate()) .setHeader(MqttHeaders.RECEIVED_RETAINED, mqttMessage.isRetained()); @@ -232,29 +259,42 @@ public class DefaultPahoMessageConverter implements MqttMessageConverter, BeanFa /** * Subclasses can override this method to convert the payload to a byte[]. * The default implementation accepts a byte[] or String payload. + * If a {@link BytesMessageMapper} is provided, conversion to byte[] + * is delegated to it, so any payload that it can handle is supported. * * @param message The outbound Message. * @return The byte[] which will become the payload of the MQTT Message. */ protected byte[] messageToMqttBytes(Message message) { - Object payload = message.getPayload(); - Assert.isTrue(payload instanceof byte[] || payload instanceof String, - () -> "This default converter can only handle 'byte[]' or 'String' payloads; consider adding a " - + "transformer to your flow definition, or subclass this converter for " - + payload.getClass().getName() + " payloads"); - byte[] payloadBytes; - if (payload instanceof String) { + if (this.bytesMessageMapper != null) { try { - payloadBytes = ((String) payload).getBytes(this.charset); + return this.bytesMessageMapper.fromMessage(message); } catch (Exception e) { - throw new MessageConversionException("failed to convert Message to object", e); + throw new MessageHandlingException(message, "Failed to map outbound message", e); } } else { - payloadBytes = (byte[]) payload; + Object payload = message.getPayload(); + Assert.isTrue(payload instanceof byte[] || payload instanceof String, + () -> "This default converter can only handle 'byte[]' or 'String' payloads; consider adding a " + + "transformer to your flow definition, or provide a BytesMessageMapper, " + + "or subclass this converter for " + + payload.getClass().getName() + " payloads"); + byte[] payloadBytes; + if (payload instanceof String) { + try { + payloadBytes = ((String) payload).getBytes(this.charset); + } + catch (Exception e) { + throw new MessageConversionException("failed to convert Message to object", e); + } + } + else { + payloadBytes = (byte[]) payload; + } + return payloadBytes; } - return payloadBytes; } } diff --git a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackToBackAdapterTests.java b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackToBackAdapterTests.java index b8f4d87dca..219e76b23d 100644 --- a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackToBackAdapterTests.java +++ b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackToBackAdapterTests.java @@ -25,6 +25,7 @@ import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; import static org.mockito.Mockito.mock; +import java.util.Collections; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -45,8 +46,11 @@ import org.springframework.integration.mqtt.event.MqttMessageDeliveredEvent; import org.springframework.integration.mqtt.event.MqttMessageSentEvent; import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter; import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler; +import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter; import org.springframework.integration.mqtt.support.MqttHeaders; import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.support.json.EmbeddedJsonHeadersMessageMapper; +import org.springframework.integration.support.json.JacksonJsonUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessagingException; @@ -104,6 +108,39 @@ public class BackToBackAdapterTests { assertEquals("mqtt-foo", out.getHeaders().get(MqttHeaders.RECEIVED_TOPIC)); } + @Test + public void testJson() { + MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("tcp://localhost:1883", "si-test-out"); + adapter.setDefaultTopic("mqtt-foo"); + adapter.setBeanFactory(mock(BeanFactory.class)); + EmbeddedJsonHeadersMessageMapper mapper = new EmbeddedJsonHeadersMessageMapper( + JacksonJsonUtils.messagingAwareMapper("org.springframework")); + DefaultPahoMessageConverter converter = new DefaultPahoMessageConverter(); + converter.setBytesMessageMapper(mapper); + adapter.setConverter(converter); + adapter.afterPropertiesSet(); + adapter.start(); + MqttPahoMessageDrivenChannelAdapter inbound = new MqttPahoMessageDrivenChannelAdapter("tcp://localhost:1883", + "si-test-in", "mqtt-foo"); + QueueChannel outputChannel = new QueueChannel(); + inbound.setOutputChannel(outputChannel); + ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); + taskScheduler.initialize(); + inbound.setTaskScheduler(taskScheduler); + inbound.setBeanFactory(mock(BeanFactory.class)); + inbound.setConverter(converter); + inbound.afterPropertiesSet(); + inbound.start(); + adapter.handleMessage(new GenericMessage(new Foo("bar"), Collections.singletonMap("baz", "qux"))); + adapter.stop(); + Message out = outputChannel.receive(10000); + assertNotNull(out); + inbound.stop(); + assertEquals(new Foo("bar"), out.getPayload()); + assertEquals("mqtt-foo", out.getHeaders().get(MqttHeaders.RECEIVED_TOPIC)); + assertEquals("qux", out.getHeaders().get("baz")); + } + @Test public void testAddRemoveTopic() { MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("tcp://localhost:1883", "si-test-out"); @@ -355,4 +392,57 @@ public class BackToBackAdapterTests { } + public static class Foo { + + private String bar; + + public Foo() { + super(); + } + + public Foo(String bar) { + this.bar = bar; + } + + public String getBar() { + return this.bar; + } + + public void setBar(String bar) { + this.bar = bar; + } + + @Override + public int hashCode() { + final int prime = 31; + int result = 1; + result = prime * result + ((this.bar == null) ? 0 : this.bar.hashCode()); + return result; + } + + @Override + public boolean equals(Object obj) { + if (this == obj) { + return true; + } + if (obj == null) { + return false; + } + if (getClass() != obj.getClass()) { + return false; + } + Foo other = (Foo) obj; + if (this.bar == null) { + if (other.bar != null) { + return false; + } + } + else if (!this.bar.equals(other.bar)) { + return false; + } + return true; + } + + } + }