diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayMethodInboundMessageMapper.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayMethodInboundMessageMapper.java index 17d9570816..91d7b60bf9 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayMethodInboundMessageMapper.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayMethodInboundMessageMapper.java @@ -45,6 +45,7 @@ import org.springframework.integration.support.AbstractIntegrationMessageBuilder import org.springframework.integration.support.DefaultMessageBuilderFactory; import org.springframework.integration.support.MessageBuilderFactory; import org.springframework.integration.util.MessagingAnnotationUtils; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessagingException; import org.springframework.messaging.core.GenericMessagingTemplate; @@ -172,19 +173,19 @@ class GatewayMethodInboundMessageMapper implements InboundMessageMapper toMessage(Object[] arguments) { + public Message toMessage(Object[] arguments, @Nullable Map headers) { Assert.notNull(arguments, "cannot map null arguments to Message"); if (arguments.length != this.parameterList.size()) { String prefix = (arguments.length < this.parameterList.size()) ? "Not enough" : "Too many"; throw new IllegalArgumentException(prefix + " parameters provided for method [" + this.method + "], expected " + this.parameterList.size() + " but received " + arguments.length + "."); } - return this.mapArgumentsToMessage(arguments); + return mapArgumentsToMessage(arguments, headers); } - private Message mapArgumentsToMessage(Object[] arguments) { + private Message mapArgumentsToMessage(Object[] arguments, Map headers) { try { - return this.argsMapper.toMessage(new MethodArgsHolder(this.method, arguments)); + return this.argsMapper.toMessage(new MethodArgsHolder(this.method, arguments), headers); } catch (Exception e) { if (e instanceof MessagingException) { @@ -276,12 +277,15 @@ class GatewayMethodInboundMessageMapper implements InboundMessageMapper toMessage(MethodArgsHolder holder) throws Exception { + public Message toMessage(MethodArgsHolder holder, @Nullable Map headers) throws Exception { Object messageOrPayload = null; boolean foundPayloadAnnotation = false; Object[] arguments = holder.getArgs(); EvaluationContext methodInvocationEvaluationContext = createMethodInvocationEvaluationContext(arguments); - Map headers = new HashMap(); + headers = + headers != null + ? new HashMap<>(headers) + : new HashMap<>(); if (GatewayMethodInboundMessageMapper.this.payloadExpression != null) { messageOrPayload = GatewayMethodInboundMessageMapper.this.payloadExpression.getValue(methodInvocationEvaluationContext); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java index 015928f3cb..0e78dee1b7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java @@ -16,6 +16,7 @@ package org.springframework.integration.gateway; +import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.atomic.AtomicLong; @@ -792,11 +793,14 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint } @Override - public Message toMessage(Object object) throws Exception { + public Message toMessage(Object object, @Nullable Map headers) throws Exception { if (object instanceof Message) { return (Message) object; } - return (object != null) ? this.messageBuilderFactory.withPayload(object).build() : null; + + return object != null + ? this.messageBuilderFactory.withPayload(object).copyHeadersIfAbsent(headers).build() + : null; } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/mapping/InboundMessageMapper.java b/spring-integration-core/src/main/java/org/springframework/integration/mapping/InboundMessageMapper.java index 95b1058a73..6b46df2df9 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/mapping/InboundMessageMapper.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/mapping/InboundMessageMapper.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. @@ -16,16 +16,39 @@ package org.springframework.integration.mapping; +import java.util.Map; + +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; /** * Strategy interface for mapping from an Object to a{@link Message}. * * @author Mark Fisher + * @author Artem Bilan */ @FunctionalInterface public interface InboundMessageMapper { - Message toMessage(T object) throws Exception; + /** + * Convert a provided object to the {@link Message}. + * @param object the object for message payload or some other conversion logic + * @return the message as a result of mapping + * @throws Exception the exception thrown by the underlying mapper implementation + */ + default Message toMessage(T object) throws Exception { + return toMessage(object, null); + } + + /** + * Convert a provided object to the {@link Message} + * and supply with headers if necessary and provided. + * @param object the object for message payload or some other conversion logic + * @param headers additional headers for building message. Can be null + * @return the message as a result of mapping + * @throws Exception the exception thrown by the underlying mapper implementation + * @since 5.0 + */ + Message toMessage(T object, @Nullable Map headers) throws Exception; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/AbstractIntegrationMessageBuilder.java b/spring-integration-core/src/main/java/org/springframework/integration/support/AbstractIntegrationMessageBuilder.java index ae5db552c9..4e431e0d23 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/AbstractIntegrationMessageBuilder.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/AbstractIntegrationMessageBuilder.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2016 the original author or authors. + * Copyright 2014-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. @@ -24,6 +24,7 @@ import java.util.List; import java.util.Map; import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHeaders; @@ -31,6 +32,8 @@ import org.springframework.util.Assert; /** * @author Gary Russell + * @author Artem Bilan + * * @since 4.0 * */ @@ -47,7 +50,7 @@ public abstract class AbstractIntegrationMessageBuilder { * @param headerValue The header value. * @return this. */ - public abstract AbstractIntegrationMessageBuilder setHeader(String headerName, Object headerValue); + public abstract AbstractIntegrationMessageBuilder setHeader(String headerName, @Nullable Object headerValue); /** * Set the value for the given header name only if the header name is not already associated with a value. @@ -86,7 +89,7 @@ public abstract class AbstractIntegrationMessageBuilder { * @see MessageHeaders#ID * @see MessageHeaders#TIMESTAMP */ - public abstract AbstractIntegrationMessageBuilder copyHeaders(Map headersToCopy); + public abstract AbstractIntegrationMessageBuilder copyHeaders(@Nullable Map headersToCopy); /** * Copy the name-value pairs from the provided Map. This operation will not overwrite any existing values. @@ -94,7 +97,7 @@ public abstract class AbstractIntegrationMessageBuilder { * @param headersToCopy The headers to copy. * @return this. */ - public abstract AbstractIntegrationMessageBuilder copyHeadersIfAbsent(Map headersToCopy); + public abstract AbstractIntegrationMessageBuilder copyHeadersIfAbsent(@Nullable Map headersToCopy); public AbstractIntegrationMessageBuilder setExpirationDate(Long expirationDate) { return this.setHeader(IntegrationMessageHeaderAccessor.EXPIRATION_DATE, expirationDate); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/MessageBuilder.java b/spring-integration-core/src/main/java/org/springframework/integration/support/MessageBuilder.java index 98d6c13bd2..70f47440ce 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/MessageBuilder.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/MessageBuilder.java @@ -24,6 +24,7 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHeaders; @@ -52,6 +53,7 @@ public final class MessageBuilder extends AbstractIntegrationMessageBuilder originalMessage; private volatile boolean modified; @@ -113,7 +115,7 @@ public final class MessageBuilder extends AbstractIntegrationMessageBuilder setHeader(String headerName, Object headerValue) { + public MessageBuilder setHeader(String headerName, @Nullable Object headerValue) { this.headerAccessor.setHeader(headerName, headerValue); return this; } @@ -173,7 +175,7 @@ public final class MessageBuilder extends AbstractIntegrationMessageBuilder copyHeaders(Map headersToCopy) { + public MessageBuilder copyHeaders(@Nullable Map headersToCopy) { this.headerAccessor.copyHeaders(headersToCopy); return this; } @@ -185,7 +187,7 @@ public final class MessageBuilder extends AbstractIntegrationMessageBuilder copyHeadersIfAbsent(Map headersToCopy) { + public MessageBuilder copyHeadersIfAbsent(@Nullable Map headersToCopy) { if (headersToCopy != null) { for (Map.Entry entry : headersToCopy.entrySet()) { String headerName = entry.getKey(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/MutableMessage.java b/spring-integration-core/src/main/java/org/springframework/integration/support/MutableMessage.java index b5de8ba803..05d01e53eb 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/MutableMessage.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/MutableMessage.java @@ -85,17 +85,16 @@ public class MutableMessage implements Message, Serializable { return this.headers.getRawHeaders(); } - @Override public String toString() { - StringBuilder sb = new StringBuilder(); + StringBuilder sb = new StringBuilder(getClass().getSimpleName()); + sb.append(" [payload="); if (this.payload instanceof byte[]) { - sb.append("[Payload byte[").append(((byte[]) this.payload).length).append("]]"); + sb.append("byte[").append(((byte[]) this.payload).length).append("]"); } else { - sb.append("[Payload ").append(this.payload.getClass().getSimpleName()); - sb.append(" content=").append(this.payload).append("]"); + sb.append(this.payload); } - sb.append("[Headers=").append(this.headers).append("]"); + sb.append(", headers=").append(this.headers).append("]"); return sb.toString(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/MutableMessageBuilder.java b/spring-integration-core/src/main/java/org/springframework/integration/support/MutableMessageBuilder.java index ed4ad51349..294a7e75c7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/MutableMessageBuilder.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/MutableMessageBuilder.java @@ -22,6 +22,7 @@ import java.util.Map; import java.util.Map.Entry; import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; import org.springframework.util.Assert; @@ -121,7 +122,7 @@ public final class MutableMessageBuilder extends AbstractIntegrationMessageBu } @Override - public AbstractIntegrationMessageBuilder setHeader(String headerName, Object headerValue) { + public AbstractIntegrationMessageBuilder setHeader(String headerName, @Nullable Object headerValue) { Assert.notNull(headerName, "'headerName' must not be null"); if (headerValue == null) { this.removeHeader(headerName); @@ -180,7 +181,7 @@ public final class MutableMessageBuilder extends AbstractIntegrationMessageBu } @Override - public AbstractIntegrationMessageBuilder copyHeaders(Map headersToCopy) { + public AbstractIntegrationMessageBuilder copyHeaders(@Nullable Map headersToCopy) { if (headersToCopy != null) { this.headers.putAll(headersToCopy); } @@ -188,7 +189,7 @@ public final class MutableMessageBuilder extends AbstractIntegrationMessageBu } @Override - public AbstractIntegrationMessageBuilder copyHeadersIfAbsent(Map headersToCopy) { + public AbstractIntegrationMessageBuilder copyHeadersIfAbsent(@Nullable Map headersToCopy) { if (headersToCopy != null) { for (Entry entry : headersToCopy.entrySet()) { setHeaderIfAbsent(entry.getKey(), entry.getValue()); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/converter/MapMessageConverter.java b/spring-integration-core/src/main/java/org/springframework/integration/support/converter/MapMessageConverter.java index a47b52791d..42a07cbbcb 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/converter/MapMessageConverter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/converter/MapMessageConverter.java @@ -26,6 +26,7 @@ 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.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.converter.MessageConverter; @@ -90,8 +91,9 @@ public class MapMessageConverter implements MessageConverter, BeanFactoryAware { this.filterHeadersInToMessage = filterHeadersInToMessage; } + @Nullable @Override - public Message toMessage(Object object, MessageHeaders messageHeaders) { + public Message toMessage(Object object, @Nullable MessageHeaders messageHeaders) { Assert.isInstanceOf(Map.class, object, "This converter expects a Map"); @SuppressWarnings("unchecked") Map map = (Map) object; @@ -106,9 +108,12 @@ public class MapMessageConverter implements MessageConverter, BeanFactoryAware { } messageBuilder.copyHeaders(headers); } - return messageBuilder.build(); + return messageBuilder + .copyHeadersIfAbsent(messageHeaders) + .build(); } + @Nullable @Override public Object fromMessage(Message message, Class clazz) { Map map = new HashMap(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/converter/SimpleMessageConverter.java b/spring-integration-core/src/main/java/org/springframework/integration/support/converter/SimpleMessageConverter.java index 8a2d958b7f..0bd8c5bef0 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/converter/SimpleMessageConverter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/converter/SimpleMessageConverter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 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. @@ -16,6 +16,8 @@ package org.springframework.integration.support.converter; +import java.util.Map; + import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; @@ -24,6 +26,7 @@ import org.springframework.integration.mapping.OutboundMessageMapper; import org.springframework.integration.support.DefaultMessageBuilderFactory; import org.springframework.integration.support.MessageBuilderFactory; import org.springframework.integration.support.utils.IntegrationUtils; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.converter.MessageConversionException; @@ -33,9 +36,10 @@ import org.springframework.messaging.converter.MessageConverter; * @author Mark Fisher * @author Gary Russell * @author Artem Bilan + * * @since 2.0 */ -@SuppressWarnings({"unchecked", "rawtypes"}) +@SuppressWarnings({ "unchecked", "rawtypes" }) public class SimpleMessageConverter implements MessageConverter, BeanFactoryAware { private volatile InboundMessageMapper inboundMessageMapper; @@ -100,16 +104,18 @@ public class SimpleMessageConverter implements MessageConverter, BeanFactoryAwar return this.messageBuilderFactory; } + @Nullable @Override - public Message toMessage(Object object, MessageHeaders headers) { + public Message toMessage(Object object, @Nullable MessageHeaders headers) { try { - return this.inboundMessageMapper.toMessage(object); + return this.inboundMessageMapper.toMessage(object, headers); } catch (Exception e) { throw new MessageConversionException("failed to convert object to Message", e); } } + @Nullable @Override public Object fromMessage(Message message, Class targetClass) { try { @@ -128,14 +134,17 @@ public class SimpleMessageConverter implements MessageConverter, BeanFactoryAwar } @Override - public Message toMessage(Object object) throws Exception { + public Message toMessage(Object object, @Nullable Map headers) throws Exception { if (object == null) { return null; } if (object instanceof Message) { return (Message) object; } - return getMessageBuilderFactory().withPayload(object).build(); + return getMessageBuilderFactory() + .withPayload(object) + .copyHeadersIfAbsent(headers) + .build(); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/json/AbstractJacksonJsonMessageParser.java b/spring-integration-core/src/main/java/org/springframework/integration/support/json/AbstractJacksonJsonMessageParser.java index dac7f81e4b..224b549ab1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/json/AbstractJacksonJsonMessageParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/json/AbstractJacksonJsonMessageParser.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. @@ -17,6 +17,7 @@ package org.springframework.integration.support.json; import java.lang.reflect.Type; +import java.util.Map; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; @@ -24,12 +25,14 @@ import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.integration.support.DefaultMessageBuilderFactory; import org.springframework.integration.support.MessageBuilderFactory; import org.springframework.integration.support.utils.IntegrationUtils; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; /** * Base {@link JsonInboundMessageMapper.JsonMessageParser} implementation for Jackson processors. * * @author Artem Bilan + * * @since 3.0 * */ @@ -66,18 +69,23 @@ abstract class AbstractJacksonJsonMessageParser

implements JsonInboundMessage } @Override - public Message doInParser(JsonInboundMessageMapper messageMapper, String jsonMessage) throws Exception { + public Message doInParser(JsonInboundMessageMapper messageMapper, String jsonMessage, + @Nullable Map headers) throws Exception { + if (this.messageMapper == null) { this.messageMapper = messageMapper; } P parser = this.createJsonParser(jsonMessage); if (messageMapper.isMapToPayload()) { - Object payload = this.readPayload(parser, jsonMessage); - return getMessageBuilderFactory().withPayload(payload).build(); + Object payload = readPayload(parser, jsonMessage); + return getMessageBuilderFactory() + .withPayload(payload) + .copyHeaders(headers) + .build(); } else { - return this.parseWithHeaders(parser, jsonMessage); + return parseWithHeaders(parser, jsonMessage, headers); } } @@ -92,8 +100,7 @@ abstract class AbstractJacksonJsonMessageParser

implements JsonInboundMessage } protected Object readHeader(P parser, String headerName, String jsonMessage) throws Exception { - Class headerType = this.messageMapper.getHeaderTypes().containsKey(headerName) ? - this.messageMapper.getHeaderTypes().get(headerName) : Object.class; + Class headerType = this.messageMapper.getHeaderTypes().getOrDefault(headerName, Object.class); try { return this.objectMapper.fromJson(parser, (Type) headerType); } @@ -103,7 +110,8 @@ abstract class AbstractJacksonJsonMessageParser

implements JsonInboundMessage } } - protected abstract Message parseWithHeaders(P parser, String jsonMessage) throws Exception; + protected abstract Message parseWithHeaders(P parser, String jsonMessage, + @Nullable Map headers) throws Exception; protected abstract P createJsonParser(String jsonMessage) throws Exception; 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 index abb2a73cbb..b42a3cb443 100644 --- 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 @@ -31,6 +31,7 @@ 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.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.support.GenericMessage; import org.springframework.util.PatternMatchUtils; @@ -70,6 +71,7 @@ import com.fasterxml.jackson.databind.ObjectMapper; * A constructor is provided allowing the provision of such a configured object mapper. * * @author Gary Russell + * @author Artem Bilan * * @since 5.0 * @@ -187,10 +189,10 @@ public class EmbeddedJsonHeadersMessageMapper implements BytesMessageMapper { } @Override - public Message toMessage(byte[] bytes) throws Exception { + public Message toMessage(byte[] bytes, @Nullable Map headers) throws Exception { Message message = null; try { - message = decodeNativeFormat(bytes); + message = decodeNativeFormat(bytes, headers); } catch (Exception e) { // empty @@ -209,11 +211,11 @@ public class EmbeddedJsonHeadersMessageMapper implements BytesMessageMapper { return message; } else { - return new GenericMessage<>(bytes); + return new GenericMessage<>(bytes, headers); } } - private Message decodeNativeFormat(byte[] bytes) throws Exception { + private Message decodeNativeFormat(byte[] bytes, Map headersToAdd) throws Exception { ByteBuffer buffer = ByteBuffer.wrap(bytes); if (buffer.remaining() > 4) { int headersLen = buffer.getInt(); @@ -228,13 +230,19 @@ public class EmbeddedJsonHeadersMessageMapper implements BytesMessageMapper { @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)); + + if (headersToAdd != null) { + headersToAdd.forEach(headers::putIfAbsent); + } + + return new GenericMessage<>(payload, new MutableMessageHeaders(headers)); } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/json/Jackson2JsonMessageParser.java b/spring-integration-core/src/main/java/org/springframework/integration/support/json/Jackson2JsonMessageParser.java index 6526c170f8..1fb6361ff4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/json/Jackson2JsonMessageParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/json/Jackson2JsonMessageParser.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. @@ -20,6 +20,7 @@ package org.springframework.integration.support.json; import java.util.LinkedHashMap; import java.util.Map; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.util.Assert; @@ -34,6 +35,7 @@ import com.fasterxml.jackson.core.JsonToken; * * @author Artem Bilan * @author Gary Russell + * * @since 3.0 */ public class Jackson2JsonMessageParser extends AbstractJacksonJsonMessageParser { @@ -52,7 +54,9 @@ public class Jackson2JsonMessageParser extends AbstractJacksonJsonMessageParser< } @Override - protected Message parseWithHeaders(JsonParser parser, String jsonMessage) throws Exception { + protected Message parseWithHeaders(JsonParser parser, String jsonMessage, + @Nullable Map headersToAdd) throws Exception { + String error = AbstractJsonInboundMessageMapper.MESSAGE_FORMAT_ERROR + jsonMessage; Assert.isTrue(JsonToken.START_OBJECT == parser.nextToken(), error); Map headers = null; @@ -72,11 +76,16 @@ public class Jackson2JsonMessageParser extends AbstractJacksonJsonMessageParser< } } Assert.notNull(headers, error); - return this.getMessageBuilderFactory().withPayload(payload).copyHeaders(headers).build(); + + return getMessageBuilderFactory() + .withPayload(payload) + .copyHeaders(headers) + .copyHeadersIfAbsent(headersToAdd) + .build(); } private Map readHeaders(JsonParser parser, String jsonMessage) throws Exception { - Map headers = new LinkedHashMap(); + Map headers = new LinkedHashMap<>(); while (JsonToken.END_OBJECT != parser.nextToken()) { String headerName = parser.getCurrentName(); parser.nextToken(); 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 3ef52c5948..58b9d4036b 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 @@ -20,6 +20,7 @@ import java.lang.reflect.Type; import java.util.Map; import org.springframework.integration.support.json.JsonInboundMessageMapper.JsonMessageParser; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.util.Assert; @@ -34,6 +35,7 @@ import org.springframework.util.Assert; * @author Mark Fisher * @author Artem Bilan * @author Gary Russell + * * @since 2.0 */ public class JsonInboundMessageMapper extends AbstractJsonInboundMessageMapper> { @@ -63,8 +65,8 @@ public class JsonInboundMessageMapper extends AbstractJsonInboundMessageMapper toMessage(String jsonMessage) throws Exception { - return this.messageParser.doInParser(this, jsonMessage); + public Message toMessage(String jsonMessage, @Nullable Map headers) throws Exception { + return this.messageParser.doInParser(this, jsonMessage, headers); } @Override @@ -81,7 +83,8 @@ public class JsonInboundMessageMapper extends AbstractJsonInboundMessageMapper { - Message doInParser(JsonInboundMessageMapper messageMapper, String jsonMessage) throws Exception; + Message doInParser(JsonInboundMessageMapper messageMapper, String jsonMessage, + @Nullable Map headers) throws Exception; } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java index 2bb417d48f..0a8ded5aab 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 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. @@ -18,6 +18,7 @@ package org.springframework.integration.config; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; import java.util.concurrent.TimeUnit; @@ -30,6 +31,7 @@ import org.springframework.messaging.support.GenericMessage; /** * @author Mark Fisher * @author Gary Russell + * @author Artem Bilan */ public class EndpointParserTests { @@ -41,8 +43,8 @@ public class EndpointParserTests { MessageChannel channel = (MessageChannel) context.getBean("endpointParserTestInput"); TestHandler handler = (TestHandler) context.getBean("testHandler"); assertNull(handler.getMessageString()); - channel.send(new GenericMessage("test")); - handler.getLatch().await(500, TimeUnit.MILLISECONDS); + channel.send(new GenericMessage<>("test")); + assertTrue(handler.getLatch().await(10000, TimeUnit.MILLISECONDS)); assertEquals("test", handler.getMessageString()); context.close(); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests.java index 7b68129d31..7a56c8d320 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests.java @@ -78,6 +78,7 @@ import org.springframework.integration.context.IntegrationProperties; import org.springframework.integration.handler.BridgeHandler; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.test.util.TestUtils; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; @@ -522,8 +523,10 @@ public class GatewayInterfaceTests { public static class BazMapper implements MethodArgsMessageMapper { @Override - public Message toMessage(MethodArgsHolder object) throws Exception { - return MessageBuilder.withPayload("fizbuz").build(); + public Message toMessage(MethodArgsHolder object, @Nullable Map headers) throws Exception { + return MessageBuilder.withPayload("fizbuz") + .copyHeadersIfAbsent(headers) + .build(); } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/MessageConvertingTcpMessageMapper.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/MessageConvertingTcpMessageMapper.java index 332c300659..4368a65d1f 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/MessageConvertingTcpMessageMapper.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/MessageConvertingTcpMessageMapper.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. @@ -16,13 +16,19 @@ package org.springframework.integration.ip.tcp.connection; -import org.springframework.integration.support.AbstractIntegrationMessageBuilder; +import java.util.Map; + +import org.springframework.integration.support.MutableMessageHeaders; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.converter.MessageConverter; import org.springframework.util.Assert; /** * @author Gary Russell + * @author Artem Bilan + * * @since 3.0 * */ @@ -36,14 +42,24 @@ public class MessageConvertingTcpMessageMapper extends TcpMessageMapper { } @Override - public Message toMessage(TcpConnection connection) throws Exception { + public Message toMessage(TcpConnection connection, @Nullable Map headers) throws Exception { Object data = connection.getPayload(); if (data != null) { - Message message = this.messageConverter.toMessage(data, null); - AbstractIntegrationMessageBuilder messageBuilder = this.getMessageBuilderFactory().fromMessage(message); - this.addStandardHeaders(connection, messageBuilder); - this.addCustomHeaders(connection, messageBuilder); - return messageBuilder.build(); + + MessageHeaders messageHeaders = new MutableMessageHeaders(null, MessageHeaders.ID_VALUE_NONE, -1L) { + + private static final long serialVersionUID = 3084692953798643018L; + + }; + + addStandardHeaders(connection, messageHeaders); + addCustomHeaders(connection, messageHeaders); + + if (headers != null) { + headers.forEach(messageHeaders::putIfAbsent); + } + + return this.messageConverter.toMessage(data, messageHeaders); } else { if (logger.isWarnEnabled()) { 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 d542e2b8c3..a5b6306c56 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 @@ -25,6 +25,7 @@ import org.apache.commons.logging.LogFactory; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; +import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.ip.IpHeaders; import org.springframework.integration.mapping.BytesMessageMapper; import org.springframework.integration.mapping.InboundMessageMapper; @@ -32,7 +33,9 @@ import org.springframework.integration.mapping.OutboundMessageMapper; import org.springframework.integration.support.AbstractIntegrationMessageBuilder; import org.springframework.integration.support.DefaultMessageBuilderFactory; import org.springframework.integration.support.MessageBuilderFactory; +import org.springframework.integration.support.MutableMessageHeaders; import org.springframework.integration.support.utils.IntegrationUtils; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.MessageHeaders; @@ -53,6 +56,7 @@ import org.springframework.util.MimeType; * * * @author Gary Russell * @author Artem Bilan + * * @since 2.0 * */ @@ -167,7 +171,7 @@ public class TcpMessageMapper implements @SuppressWarnings("unchecked") @Override - public Message toMessage(TcpConnection connection) throws Exception { + public Message toMessage(TcpConnection connection, @Nullable Map headers) throws Exception { Message message = null; Object payload = connection.getPayload(); if (payload != null) { @@ -177,11 +181,19 @@ public class TcpMessageMapper implements .fromMessage(this.bytesMessageMapper.toMessage((byte[]) payload)); } else { - messageBuilder = getMessageBuilderFactory().withPayload(payload); + messageBuilder = getMessageBuilderFactory() + .withPayload(payload); } - this.addStandardHeaders(connection, messageBuilder); - this.addCustomHeaders(connection, messageBuilder); - message = messageBuilder.build(); + + MessageHeaders messageHeaders = new MutableMessageHeaders(null); + + addStandardHeaders(connection, messageHeaders); + addCustomHeaders(connection, messageHeaders); + + message = messageBuilder + .copyHeaders(messageHeaders) + .copyHeadersIfAbsent(headers) + .build(); } else { if (this.logger.isWarnEnabled()) { @@ -191,33 +203,32 @@ public class TcpMessageMapper implements return message; } - protected final void addStandardHeaders(TcpConnection connection, - AbstractIntegrationMessageBuilder messageBuilder) { + protected final void addStandardHeaders(TcpConnection connection, MessageHeaders messageHeaders) { String connectionId = connection.getConnectionId(); - messageBuilder - .setHeader(IpHeaders.HOSTNAME, connection.getHostName()) - .setHeader(IpHeaders.IP_ADDRESS, connection.getHostAddress()) - .setHeader(IpHeaders.REMOTE_PORT, connection.getPort()) - .setHeader(IpHeaders.CONNECTION_ID, connectionId); + + messageHeaders.put(IpHeaders.HOSTNAME, connection.getHostName()); + messageHeaders.put(IpHeaders.IP_ADDRESS, connection.getHostAddress()); + messageHeaders.put(IpHeaders.REMOTE_PORT, connection.getPort()); + messageHeaders.put(IpHeaders.CONNECTION_ID, connectionId); + SocketInfo socketInfo = connection.getSocketInfo(); if (socketInfo != null) { - messageBuilder.setHeader(IpHeaders.LOCAL_ADDRESS, socketInfo.getLocalAddress()); + messageHeaders.put(IpHeaders.LOCAL_ADDRESS, socketInfo.getLocalAddress()); } if (this.applySequence) { - messageBuilder - .setCorrelationId(connectionId) - .setSequenceNumber((int) connection.incrementAndGetConnectionSequence()); + messageHeaders.put(IntegrationMessageHeaderAccessor.CORRELATION_ID, connectionId); + messageHeaders.put(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, + connection.incrementAndGetConnectionSequence()); } if (this.addContentTypeHeader) { - messageBuilder.setHeader(MessageHeaders.CONTENT_TYPE, this.contentType); + messageHeaders.put(MessageHeaders.CONTENT_TYPE, this.contentType); } } - protected final void addCustomHeaders(TcpConnection connection, - AbstractIntegrationMessageBuilder messageBuilder) { - Map customHeaders = this.supplyCustomHeaders(connection); + protected final void addCustomHeaders(TcpConnection connection, MessageHeaders messageHeaders) { + Map customHeaders = supplyCustomHeaders(connection); if (customHeaders != null) { - messageBuilder.copyHeadersIfAbsent(customHeaders); + customHeaders.forEach(messageHeaders::putIfAbsent); } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapper.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapper.java index 5096990188..f26deccaa6 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapper.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapper.java @@ -19,6 +19,7 @@ package org.springframework.integration.ip.udp; import java.io.UnsupportedEncodingException; import java.net.DatagramPacket; import java.nio.ByteBuffer; +import java.util.Map; import java.util.UUID; import java.util.regex.Matcher; import java.util.regex.Pattern; @@ -34,6 +35,7 @@ import org.springframework.integration.mapping.OutboundMessageMapper; import org.springframework.integration.support.DefaultMessageBuilderFactory; import org.springframework.integration.support.MessageBuilderFactory; import org.springframework.integration.support.utils.IntegrationUtils; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.MessageHeaders; @@ -61,6 +63,7 @@ import org.springframework.util.StringUtils; * @author Gary Russell * @author Dave Syer * @author Artem Bilan + * * @since 2.0 */ public class DatagramPacketMessageMapper implements InboundMessageMapper, @@ -196,7 +199,12 @@ public class DatagramPacketMessageMapper implements InboundMessageMapper toMessage(DatagramPacket packet) throws Exception { + public Message toMessage(DatagramPacket object) throws Exception { + return toMessage(object, null); + } + + @Override + public Message toMessage(DatagramPacket packet, @Nullable Map headers) throws Exception { int offset = packet.getOffset(); int length = packet.getLength(); byte[] payload; @@ -205,7 +213,8 @@ public class DatagramPacketMessageMapper implements InboundMessageMapper convertMessage(Object object) { - return this.messageConverter.toMessage(object, null); + private Message convertMessage(Object object, String source) { + MessageHeaders messageHeaders = null; + if (StringUtils.hasText(source)) { + messageHeaders = new MessageHeaders(Collections.singletonMap(RedisHeaders.MESSAGE_SOURCE, source)); + } + + return this.messageConverter.toMessage(object, messageHeaders); } @@ -150,9 +165,10 @@ public class RedisInboundChannelAdapter extends MessageProducerSupport { } @SuppressWarnings("unused") - public void handleMessage(Object object) { - sendMessage(convertMessage(object)); + public void handleMessage(Object message, String source) { + sendMessage(convertMessage(message, source)); } + } } diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/support/RedisHeaders.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/support/RedisHeaders.java index 2e09353322..ea9c49c182 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/support/RedisHeaders.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/support/RedisHeaders.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2016 the original author or authors. + * Copyright 2014-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. @@ -23,6 +23,7 @@ package org.springframework.integration.redis.support; * @author Oleg Zhurakousky * @author Gary Russell * @author Artem Bilan + * * @since 2.2 */ public final class RedisHeaders { @@ -43,4 +44,6 @@ public final class RedisHeaders { public static final String COMMAND = PREFIX + "command"; + public static final String MESSAGE_SOURCE = PREFIX + "messageSource"; + } diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapterTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapterTests.java index b0678fcbc2..1506b66ff3 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapterTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapterTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2007-2016 the original author or authors. + * Copyright 2007-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. @@ -16,10 +16,10 @@ package org.springframework.integration.redis.inbound; +import static org.hamcrest.Matchers.startsWith; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertThat; -import static org.junit.Assert.assertTrue; import static org.mockito.Mockito.mock; import org.hamcrest.Matchers; @@ -33,6 +33,7 @@ import org.springframework.data.redis.listener.RedisMessageListenerContainer; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.redis.rules.RedisAvailable; import org.springframework.integration.redis.rules.RedisAvailableTests; +import org.springframework.integration.redis.support.RedisHeaders; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; @@ -40,6 +41,7 @@ import org.springframework.messaging.Message; * @author Mark Fisher * @author Artem Bilan * @author Gary Russell + * * @since 2.1 */ public class RedisInboundChannelAdapterTests extends RedisAvailableTests { @@ -83,7 +85,8 @@ public class RedisInboundChannelAdapterTests extends RedisAvailableTests { throw new RuntimeException("Failed to receive message # " + i + " iteration " + iteration); } assertNotNull(message); - assertTrue(message.getPayload().toString().startsWith("test-")); + assertThat(message.getPayload().toString(), startsWith("test-")); + assertEquals("testRedisInboundChannelAdapterChannel", message.getHeaders().get(RedisHeaders.MESSAGE_SOURCE)); counter++; } assertEquals(numToTest, counter); @@ -119,7 +122,7 @@ public class RedisInboundChannelAdapterTests extends RedisAvailableTests { Object payload = message.getPayload(); assertThat(payload, Matchers.instanceOf(byte[].class)); - assertTrue(new String((byte[]) payload).startsWith("test-")); + assertThat(new String((byte[]) payload), startsWith("test-")); counter++; } diff --git a/src/reference/asciidoc/redis.adoc b/src/reference/asciidoc/redis.adoc index fe48fd7860..3ffa853015 100644 --- a/src/reference/asciidoc/redis.adoc +++ b/src/reference/asciidoc/redis.adoc @@ -103,8 +103,8 @@ Instead those Messages are passed through Redis allowing you to rely on its supp [[redis-inbound-channel-adapter]] ==== Redis Inbound Channel Adapter -The Redis-based Inbound Channel Adapter adapts incoming Redis messages into Spring Integration Messages in the same way as other inbound adapters. -It receives platform-specific messages (Redis in this case) and converts them to Spring Integration Messages using a `MessageConverter` strategy. +The Redis-based Inbound Channel Adapter (`RedisInboundChannelAdapter`) adapts incoming Redis messages into Spring Messages in the same way as other inbound adapters. +It receives platform-specific messages (Redis in this case) and converts them to Spring Messages using a `MessageConverter` strategy. [source,xml] ---- ` can be s In this case the raw `byte[]` bodies of Redis Messages are provided as the message payloads. Since _version 5.0_, an `Executor` instance can be provided to the Inbound Adapter via the `task-executor` attribute of the ``. +Also the received Spring Integration Messages have now `RedisHeaders.MESSAGE_SOURCE` header to indicate the source of the published message - topic or pattern. +This can be used downstream for routing logic. [[redis-outbound-channel-adapter]] ==== Redis Outbound Channel Adapter diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 447cb61c1b..23566350be 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -273,6 +273,7 @@ The `zsetIncrementExpression` can now be configured on the `RedisStoreWritingMes In addition this property has been changed from `true` to `false` since `INCR` option on `ZADD` Redis command is optional. The `RedisInboundChannelAdapter` can now be supplied with an `Executor` for executing Redis listener invokers. +In addition the received messages now contains a `RedisHeaders.MESSAGE_SOURCE` header to indicate the source of the message - topic or pattern. See <> for more information.