Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2268: Add RedisHeaders.MESSAGE_SOURCE header

Browse Source 
Resolves: /spring-projects/spring-integration#2268

* To indicate the source the Redis message in the `RedisInboundChannelAdapter`
populate the `RedisHeaders.MESSAGE_SOURCE` header to the messages to produce
* Fix the `SimpleMessageConverter` to populate the provided `MessageHeaders`
to the message to produce

* Add `toMessage(T object, @Nullable Map<String, Object> headers)`
to the `InboundMessageMapper` to propagate additional header
to the message to create
* Rework all the out-of-the-box `InboundMessageMapper` implementations
to properly propagate additional headers via `toMessage()`
from the `SimpleMessageConverter`
* Provide optimizations in the `InboundMessageMapper` implementations
do not re-create messages
* Refactor `MutableMessage.toString()` to align with the `GenericMessage`

* Add more `@Nullable` to method arguments
* Increase latch wait timeout in the `EndpointParserTests`
* Address `redis.adoc` PR comment
This commit is contained in:
Artem Bilan
2017-11-07 13:25:33 -05:00
committed by Gary Russell
parent 70c166fbfd
commit 7ca20e53f7
23 changed files with 249 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayMethodInboundMessageMapper.java
View File

@@ -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<Object[]
}
@Override
public Message<?> toMessage(Object[] arguments) {
public Message<?> toMessage(Object[] arguments, @Nullable Map<String, Object> 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<String, Object> 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<Object[]
public class DefaultMethodArgsMessageMapper implements MethodArgsMessageMapper {
@Override
public Message<?> toMessage(MethodArgsHolder holder) throws Exception {
public Message<?> toMessage(MethodArgsHolder holder, @Nullable Map<String, Object> headers) throws Exception {
Object messageOrPayload = null;
boolean foundPayloadAnnotation = false;
Object[] arguments = holder.getArgs();
EvaluationContext methodInvocationEvaluationContext = createMethodInvocationEvaluationContext(arguments);
Map<String, Object> headers = new HashMap<String, Object>();
headers =
headers != null
? new HashMap<>(headers)
: new HashMap<>();
if (GatewayMethodInboundMessageMapper.this.payloadExpression != null) {
messageOrPayload =
GatewayMethodInboundMessageMapper.this.payloadExpression.getValue(methodInvocationEvaluationContext);

8
spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java
View File

@@ -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<String, Object> 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;
}
}

27
spring-integration-core/src/main/java/org/springframework/integration/mapping/InboundMessageMapper.java
View File

@@ -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<T> {
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<String, Object> headers) throws Exception;
}

11
spring-integration-core/src/main/java/org/springframework/integration/support/AbstractIntegrationMessageBuilder.java
View File

@@ -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<T> {
* @param headerValue The header value.
* @return this.
*/
public abstract AbstractIntegrationMessageBuilder<T> setHeader(String headerName, Object headerValue);
public abstract AbstractIntegrationMessageBuilder<T> 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<T> {
* @see MessageHeaders#ID
* @see MessageHeaders#TIMESTAMP
*/
public abstract AbstractIntegrationMessageBuilder<T> copyHeaders(Map<String, ?> headersToCopy);
public abstract AbstractIntegrationMessageBuilder<T> copyHeaders(@Nullable Map<String, ?> headersToCopy);
/**
* Copy the name-value pairs from the provided Map. This operation will <em>not</em> overwrite any existing values.
@@ -94,7 +97,7 @@ public abstract class AbstractIntegrationMessageBuilder<T> {
* @param headersToCopy The headers to copy.
* @return this.
*/
public abstract AbstractIntegrationMessageBuilder<T> copyHeadersIfAbsent(Map<String, ?> headersToCopy);
public abstract AbstractIntegrationMessageBuilder<T> copyHeadersIfAbsent(@Nullable Map<String, ?> headersToCopy);
public AbstractIntegrationMessageBuilder<T> setExpirationDate(Long expirationDate) {
return this.setHeader(IntegrationMessageHeaderAccessor.EXPIRATION_DATE, expirationDate);

8
spring-integration-core/src/main/java/org/springframework/integration/support/MessageBuilder.java
View File

@@ -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<T> extends AbstractIntegrationMessageBuilder<T
private final IntegrationMessageHeaderAccessor headerAccessor;
@Nullable
private final Message<T> originalMessage;
private volatile boolean modified;
@@ -113,7 +115,7 @@ public final class MessageBuilder<T> extends AbstractIntegrationMessageBuilder<T
* @return this MessageBuilder.
*/
@Override
public MessageBuilder<T> setHeader(String headerName, Object headerValue) {
public MessageBuilder<T> setHeader(String headerName, @Nullable Object headerValue) {
this.headerAccessor.setHeader(headerName, headerValue);
return this;
}
@@ -173,7 +175,7 @@ public final class MessageBuilder<T> extends AbstractIntegrationMessageBuilder<T
* @see MessageHeaders#TIMESTAMP
*/
@Override
public MessageBuilder<T> copyHeaders(Map<String, ?> headersToCopy) {
public MessageBuilder<T> copyHeaders(@Nullable Map<String, ?> headersToCopy) {
this.headerAccessor.copyHeaders(headersToCopy);
return this;
}
@@ -185,7 +187,7 @@ public final class MessageBuilder<T> extends AbstractIntegrationMessageBuilder<T
* @return this MessageBuilder.
*/
@Override
public MessageBuilder<T> copyHeadersIfAbsent(Map<String, ?> headersToCopy) {
public MessageBuilder<T> copyHeadersIfAbsent(@Nullable Map<String, ?> headersToCopy) {
if (headersToCopy != null) {
for (Map.Entry<String, ?> entry : headersToCopy.entrySet()) {
String headerName = entry.getKey();

11
spring-integration-core/src/main/java/org/springframework/integration/support/MutableMessage.java
View File

@@ -85,17 +85,16 @@ public class MutableMessage<T> implements Message<T>, 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();
}

7
spring-integration-core/src/main/java/org/springframework/integration/support/MutableMessageBuilder.java
View File

@@ -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<T> extends AbstractIntegrationMessageBu
}
@Override
public AbstractIntegrationMessageBuilder<T> setHeader(String headerName, Object headerValue) {
public AbstractIntegrationMessageBuilder<T> 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<T> extends AbstractIntegrationMessageBu
}
@Override
public AbstractIntegrationMessageBuilder<T> copyHeaders(Map<String, ?> headersToCopy) {
public AbstractIntegrationMessageBuilder<T> copyHeaders(@Nullable Map<String, ?> headersToCopy) {
if (headersToCopy != null) {
this.headers.putAll(headersToCopy);
}
@@ -188,7 +189,7 @@ public final class MutableMessageBuilder<T> extends AbstractIntegrationMessageBu
}
@Override
public AbstractIntegrationMessageBuilder<T> copyHeadersIfAbsent(Map<String, ?> headersToCopy) {
public AbstractIntegrationMessageBuilder<T> copyHeadersIfAbsent(@Nullable Map<String, ?> headersToCopy) {
if (headersToCopy != null) {
for (Entry<String, ?> entry : headersToCopy.entrySet()) {
setHeaderIfAbsent(entry.getKey(), entry.getValue());

9
spring-integration-core/src/main/java/org/springframework/integration/support/converter/MapMessageConverter.java
View File

@@ -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<String, ?> map = (Map<String, ?>) 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<String, Object> map = new HashMap<String, Object>();

21
spring-integration-core/src/main/java/org/springframework/integration/support/converter/SimpleMessageConverter.java
View File

@@ -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<String, Object> 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();
}
}

24
spring-integration-core/src/main/java/org/springframework/integration/support/json/AbstractJacksonJsonMessageParser.java
View File

@@ -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<P> implements JsonInboundMessage
}
@Override
public Message<?> doInParser(JsonInboundMessageMapper messageMapper, String jsonMessage) throws Exception {
public Message<?> doInParser(JsonInboundMessageMapper messageMapper, String jsonMessage,
@Nullable Map<String, Object> 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<P> 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<P> implements JsonInboundMessage
}
}
protected abstract Message<?> parseWithHeaders(P parser, String jsonMessage) throws Exception;
protected abstract Message<?> parseWithHeaders(P parser, String jsonMessage,
@Nullable Map<String, Object> headers) throws Exception;
protected abstract P createJsonParser(String jsonMessage) throws Exception;

18
spring-integration-core/src/main/java/org/springframework/integration/support/json/EmbeddedJsonHeadersMessageMapper.java
View File

@@ -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<String, Object> 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<String, Object> 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<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));
if (headersToAdd != null) {
headersToAdd.forEach(headers::putIfAbsent);
}
return new GenericMessage<>(payload, new MutableMessageHeaders(headers));
}
}
}

17
spring-integration-core/src/main/java/org/springframework/integration/support/json/Jackson2JsonMessageParser.java
View File

@@ -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<JsonParser> {
@@ -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<String, Object> headersToAdd) throws Exception {
String error = AbstractJsonInboundMessageMapper.MESSAGE_FORMAT_ERROR + jsonMessage;
Assert.isTrue(JsonToken.START_OBJECT == parser.nextToken(), error);
Map<String, Object> 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<String, Object> readHeaders(JsonParser parser, String jsonMessage) throws Exception {
Map<String, Object> headers = new LinkedHashMap<String, Object>();
Map<String, Object> headers = new LinkedHashMap<>();
while (JsonToken.END_OBJECT != parser.nextToken()) {
String headerName = parser.getCurrentName();
parser.nextToken();

9
spring-integration-core/src/main/java/org/springframework/integration/support/json/JsonInboundMessageMapper.java
View File

@@ -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<JsonMessageParser<?>> {
@@ -63,8 +65,8 @@ public class JsonInboundMessageMapper extends AbstractJsonInboundMessageMapper<J
}
@Override
public Message<?> toMessage(String jsonMessage) throws Exception {
return this.messageParser.doInParser(this, jsonMessage);
public Message<?> toMessage(String jsonMessage, @Nullable Map<String, Object> headers) throws Exception {
return this.messageParser.doInParser(this, jsonMessage, headers);
}
@Override
@@ -81,7 +83,8 @@ public class JsonInboundMessageMapper extends AbstractJsonInboundMessageMapper<J
public interface JsonMessageParser<P> {
Message<?> doInParser(JsonInboundMessageMapper messageMapper, String jsonMessage) throws Exception;
Message<?> doInParser(JsonInboundMessageMapper messageMapper, String jsonMessage,
@Nullable Map<String, Object> headers) throws Exception;
}

8
spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java
View File

@@ -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<String>("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();
}

7
spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests.java
View File

@@ -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<String, Object> headers) throws Exception {
return MessageBuilder.withPayload("fizbuz")
.copyHeadersIfAbsent(headers)
.build();
}
}

32
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/MessageConvertingTcpMessageMapper.java
View File

@@ -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<String, Object> 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()) {

53
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapper.java
View File

@@ -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<String, Object> headers) throws Exception {
Message<Object> 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<String, ?> customHeaders = this.supplyCustomHeaders(connection);
protected final void addCustomHeaders(TcpConnection connection, MessageHeaders messageHeaders) {
Map<String, ?> customHeaders = supplyCustomHeaders(connection);
if (customHeaders != null) {
messageBuilder.copyHeadersIfAbsent(customHeaders);
customHeaders.forEach(messageHeaders::putIfAbsent);
}
}

18
spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapper.java
View File

@@ -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<DatagramPacket>,
@@ -196,7 +199,12 @@ public class DatagramPacketMessageMapper implements InboundMessageMapper<Datagra
}
@Override
public Message<byte[]> toMessage(DatagramPacket packet) throws Exception {
public Message<byte[]> toMessage(DatagramPacket object) throws Exception {
return toMessage(object, null);
}
@Override
public Message<byte[]> toMessage(DatagramPacket packet, @Nullable Map<String, Object> headers) throws Exception {
int offset = packet.getOffset();
int length = packet.getLength();
byte[] payload;
@@ -205,7 +213,8 @@ public class DatagramPacketMessageMapper implements InboundMessageMapper<Datagra
if (this.lengthCheck) {
int declaredLength = buffer.getInt();
if (declaredLength != (length - 4)) {
throw new MessageMappingException("Incorrect length; expected " + (declaredLength + 4) + ", received " + length);
throw new MessageMappingException("Incorrect length; expected " + (declaredLength + 4)
+ ", received " + length);
}
offset += 4;
length -= 4;
@@ -223,8 +232,8 @@ public class DatagramPacketMessageMapper implements InboundMessageMapper<Datagra
// side expects it.
if (this.acknowledge || startsWith(buffer, IpHeaders.ACK_ADDRESS)) {
try {
String headers = new String(packet.getData(), offset, length, this.charset);
Matcher matcher = udpHeadersPattern.matcher(headers);
String headersString = new String(packet.getData(), offset, length, this.charset);
Matcher matcher = udpHeadersPattern.matcher(headersString);
if (matcher.find()) {
// Strip off the ack headers and put in Message headers
length = length - matcher.end();
@@ -237,6 +246,7 @@ public class DatagramPacketMessageMapper implements InboundMessageMapper<Datagra
.setHeader(IpHeaders.IP_ADDRESS, hostAddress)
.setHeader(IpHeaders.PORT, port)
.setHeader(IpHeaders.PACKET_ADDRESS, packet.getSocketAddress())
.copyHeadersIfAbsent(headers)
.build();
} // on no match, just treat as simple payload
}

26
spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapter.java
View File

@@ -17,6 +17,7 @@
package org.springframework.integration.redis.inbound;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.Executor;
@@ -30,10 +31,13 @@ import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.support.converter.SimpleMessageConverter;
import org.springframework.integration.redis.support.RedisHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.converter.SimpleMessageConverter;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* @author Mark Fisher
@@ -78,6 +82,12 @@ public class RedisInboundChannelAdapter extends MessageProducerSupport {
this.messageConverter = messageConverter;
}
/**
* Specify an {@link Executor} used for running the message listeners when messages are received.
* @param taskExecutor the Executor to use for listener container.
* @since 4.3.13
* @see RedisMessageListenerContainer#setTaskExecutor(Executor)
*/
public void setTaskExecutor(Executor taskExecutor) {
this.container.setTaskExecutor(taskExecutor);
}
@@ -138,8 +148,13 @@ public class RedisInboundChannelAdapter extends MessageProducerSupport {
this.container.stop();
}
private Message<?> 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));
}
}
}

5
spring-integration-redis/src/main/java/org/springframework/integration/redis/support/RedisHeaders.java
View File

@@ -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";
}

11
spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapterTests.java
View File

@@ -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++;
}

6
src/reference/asciidoc/redis.adoc
View File

@@ -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]
----
<int-redis:inbound-channel-adapter id="redisAdapter"
@@ -141,6 +141,8 @@ The `serializer` attribute of the `<int-redis:inbound-channel-adapter>` 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 `<int-redis:inbound-channel-adapter>`.
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

1
src/reference/asciidoc/whats-new.adoc
View File

@@ -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 <<redis>> for more information.