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

@@ -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'

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<byte[]>, OutboundMessageMapper<byte[]> {
+
+}

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.
+ * <p>
+ * By default, all headers are included; you can provide simple patterns to specify a
+ * subset of headers.
+ * <p>
+ * If neither expected format is detected, or an error occurs during conversion, the
+ * payload of the message is the original {@code byte[]}.
+ * <p>
+ * <b>IMPORTANT</b>
+ * <p>
+ * The default object mapper will only deserialize classes in certain packages.
+ *
+ * <pre class=code>
+ *	"java.util",
+ *	"java.lang",
+ *	"org.springframework.messaging.support",
+ *	"org.springframework.integration.support",
+ *	"org.springframework.integration.message",
+ *	"org.springframework.integration.store"
+ * </pre>
+ * <p>
+ * To add more packages, create an object mapper using
+ * {@link JacksonJsonUtils#messagingAwareMapper(String...)}.
+ * <p>
+ * 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<String> 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 &lt;headersLen&gt;&lt;headers&gt;&lt;payloadLen&gt;&lt;payload&gt;; 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<String> 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<byte[]>) messageToEncode);
+		}
+		else {
+			return this.objectMapper.writeValueAsBytes(messageToEncode);
+		}
+	}
+
+	private Message<?> pruneHeaders(Message<?> message) {
+		Map<String, Object> 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<byte[]> 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<String, Object> 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<Object>(payload, new MutableMessageHeaders(headers));
+				}
+			}
+		}
+		return null;
+	}
+
+}

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.
+ * <p>
+ * Consider using the {@link EmbeddedJsonHeadersMessageMapper} instead.
  *
  * @author Jeremy Grelle
  * @author Oleg Zhurakousky
@@ -60,6 +62,7 @@ public class JsonInboundMessageMapper extends AbstractJsonInboundMessageMapper<J
 		return headerTypes;
 	}
 
+	@Override
 	public Message<?> toMessage(String jsonMessage) throws Exception {
 		return this.messageParser.doInParser(this, jsonMessage);
 	}

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.
+ * <p>
+ * 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<String> {
@@ -47,6 +52,7 @@ public class JsonOutboundMessageMapper implements OutboundMessageMapper<String>
 		this.shouldExtractPayload = shouldExtractPayload;
 	}
 
+	@Override
 	public String fromMessage(Message<?> message) throws Exception {
 		return this.jsonObjectMapper.toJson(this.shouldExtractPayload ? message.getPayload() : message);
 	}

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<String> message = new GenericMessage<>("foo");
+		assertThat(mapper.toMessage(mapper.fromMessage(message)), equalTo(message));
+
+	}
+
+	@Test
+	public void testEmbedSome() throws Exception {
+		EmbeddedJsonHeadersMessageMapper mapper = new EmbeddedJsonHeadersMessageMapper("id");
+		GenericMessage<String> 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<byte[]> 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<byte[]> 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<byte[]> message = new GenericMessage<>("foo".getBytes());
+		byte[] mappedBytes = mapper.fromMessage(message);
+		String mapped = new String(mappedBytes);
+		assertThat(mapped, containsString("[B\",\"Zm9v"));
+		@SuppressWarnings("unchecked")
+		Message<byte[]> decoded = (Message<byte[]>) mapper.toMessage(mappedBytes);
+		assertThat(new String(decoded.getPayload()), equalTo("foo"));
+
+	}
+
+	@Test
+	public void testBytesDecodeAll() throws Exception {
+		EmbeddedJsonHeadersMessageMapper mapper = new EmbeddedJsonHeadersMessageMapper();
+		GenericMessage<byte[]> message = new GenericMessage<>("foo".getBytes());
+		Message<?> decoded = mapper.toMessage(mapper.fromMessage(message));
+		assertThat(decoded, equalTo(message));
+	}
+
+}

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<Object> message = null;
 		Object payload = connection.getPayload();
 		if (payload != null) {
-			AbstractIntegrationMessageBuilder<Object> messageBuilder = getMessageBuilderFactory().withPayload(payload);
+			AbstractIntegrationMessageBuilder<Object> messageBuilder;
+			if (this.bytesMessageMapper != null && payload instanceof byte[]) {
+				messageBuilder = (AbstractIntegrationMessageBuilder<Object>) 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);
 		}

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<String> 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));
+	}
+
 }

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<Boolean> 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<Object> messageBuilder = getMessageBuilderFactory()
-					.withPayload(mqttBytesToPayload(mqttMessage))
+			AbstractIntegrationMessageBuilder<Object> messageBuilder;
+			if (this.bytesMessageMapper != null) {
+				messageBuilder = (AbstractIntegrationMessageBuilder<Object>) 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;
 	}
 
 }

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<Foo>(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;
+		}
+
+	}
+
 }