Sonar Fixes

- avoid parameter assignments

spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2018 the original author or authors.
+ * Copyright 2002-2019 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.
@@ -407,16 +407,17 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
 	 * is interrupted. If the specified timeout is 0, the method will return
 	 * immediately. If less than zero, it will block indefinitely (see
 	 * {@link #send(Message)}).
-	 * @param message the Message to send
+	 * @param messageArg the Message to send
 	 * @param timeout the timeout in milliseconds
 	 * @return <code>true</code> if the message is sent successfully,
 	 * <code>false</code> if the message cannot be sent within the allotted
 	 * time or the sending thread is interrupted.
 	 */
 	@Override
-	public boolean send(Message<?> message, long timeout) {
-		Assert.notNull(message, "message must not be null");
-		Assert.notNull(message.getPayload(), "message payload must not be null");
+	public boolean send(Message<?> messageArg, long timeout) {
+		Assert.notNull(messageArg, "message must not be null");
+		Assert.notNull(messageArg.getPayload(), "message payload must not be null");
+		Message<?> message = messageArg;
 		if (this.shouldTrack) {
 			message = MessageHistory.write(message, this, this.getMessageBuilderFactory());
 		}
@@ -596,8 +597,10 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
 		}
 
 		@Nullable
-		public Message<?> preSend(Message<?> message, MessageChannel channel,
+		public Message<?> preSend(Message<?> messageArg, MessageChannel channel,
 				Deque<ChannelInterceptor> interceptorStack) {
+
+			Message<?> message = messageArg;
 			if (this.size > 0) {
 				for (ChannelInterceptor interceptor : this.interceptors) {
 					Message<?> previous = message;
@@ -652,7 +655,8 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
 		}
 
 		@Nullable
-		public Message<?> postReceive(Message<?> message, MessageChannel channel) {
+		public Message<?> postReceive(Message<?> messageArg, MessageChannel channel) {
+			Message<?> message = messageArg;
 			if (this.size > 0) {
 				for (ChannelInterceptor interceptor : this.interceptors) {
 					message = interceptor.postReceive(message, channel);

spring-integration-core/src/main/java/org/springframework/integration/channel/PriorityChannel.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2019 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.
@@ -124,9 +124,11 @@ public class PriorityChannel extends QueueChannel {
 			return false;
 		}
 		if (!this.useMessageStore) {
-			message = new MessageWrapper(message);
+			return super.doSend(new MessageWrapper(message), 0);
+		}
+		else {
+			return super.doSend(message, 0);
 		}
-		return super.doSend(message, 0);
 	}
 
 	@Override

spring-integration-core/src/main/java/org/springframework/integration/config/annotation/InboundChannelAdapterAnnotationPostProcessor.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2018 the original author or authors.
+ * Copyright 2014-2019 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.
@@ -100,8 +100,10 @@ public class InboundChannelAdapterAnnotationPostProcessor extends
 		return adapter;
 	}
 
-	private MessageSource<?> createMessageSource(Object bean, String beanName, Method method) {
+	private MessageSource<?> createMessageSource(Object beanArg, String beanName, Method methodArg) {
 		MessageSource<?> messageSource = null;
+		Object bean = beanArg;
+		Method method = methodArg;
 		if (AnnotatedElementUtils.isAnnotated(method, Bean.class.getName())) {
 			Object target = this.resolveTargetBeanFromMethodWithBeanAnnotation(method);
 			Class<?> targetClass = target.getClass();

spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractChannelParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2019 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.
@@ -36,6 +36,7 @@ import org.springframework.util.xml.DomUtils;
  * @author Mark Fisher
  * @author Dave Syer
  * @author Artem Bilan
+ * @author Gary Russell
  */
 public abstract class AbstractChannelParser extends AbstractBeanDefinitionParser {
 
@@ -86,10 +87,13 @@ public abstract class AbstractChannelParser extends AbstractBeanDefinitionParser
 	@Override
 	protected void registerBeanDefinition(BeanDefinitionHolder definition, BeanDefinitionRegistry registry) {
 		String scope = definition.getBeanDefinition().getScope();
-		if (!AbstractBeanDefinition.SCOPE_DEFAULT.equals(scope) && !AbstractBeanDefinition.SCOPE_SINGLETON.equals(scope) && !AbstractBeanDefinition.SCOPE_PROTOTYPE.equals(scope)) {
-			definition = ScopedProxyUtils.createScopedProxy(definition, registry, false);
+		if (!AbstractBeanDefinition.SCOPE_DEFAULT.equals(scope) && !AbstractBeanDefinition.SCOPE_SINGLETON.equals(scope)
+				&& !AbstractBeanDefinition.SCOPE_PROTOTYPE.equals(scope)) {
+			super.registerBeanDefinition(ScopedProxyUtils.createScopedProxy(definition, registry, false), registry);
+		}
+		else {
+			super.registerBeanDefinition(definition, registry);
 		}
-		super.registerBeanDefinition(definition, registry);
 	}
 
 	/**

spring-integration-core/src/main/java/org/springframework/integration/config/xml/HeaderEnricherParserSupport.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2018 the original author or authors.
+ * Copyright 2002-2019 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.
@@ -39,6 +39,7 @@ import org.springframework.integration.transformer.support.ExpressionEvaluatingH
 import org.springframework.integration.transformer.support.MessageProcessingHeaderValueMessageProcessor;
 import org.springframework.integration.transformer.support.RoutingSlipHeaderValueMessageProcessor;
 import org.springframework.integration.transformer.support.StaticHeaderValueMessageProcessor;
+import org.springframework.lang.Nullable;
 import org.springframework.util.StringUtils;
 import org.springframework.util.xml.DomUtils;
 
@@ -148,11 +149,13 @@ public abstract class HeaderEnricherParserSupport extends AbstractTransformerPar
 	}
 
 	private void addHeader(Element element, ManagedMap<String, Object> headers, ParserContext parserContext,
-			String headerName, Element headerElement, String headerType, String expression, String overwrite) {
+			String headerName, Element headerElement, String headerType, @Nullable String expressionArg,
+			String overwrite) {
 
 		String value = headerElement.getAttribute("value");
 		String ref = headerElement.getAttribute(REF_ATTRIBUTE);
 		String method = headerElement.getAttribute(METHOD_ATTRIBUTE);
+		String expression = expressionArg;
 		if (expression == null) {
 			expression = headerElement.getAttribute(EXPRESSION_ATTRIBUTE);
 		}

spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2018 the original author or authors.
+ * Copyright 2002-2019 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.
@@ -47,6 +47,7 @@ import org.springframework.integration.config.FixedSubscriberChannelBeanFactoryP
 import org.springframework.integration.config.IntegrationConfigUtils;
 import org.springframework.integration.context.IntegrationContextUtils;
 import org.springframework.integration.transaction.TransactionHandleMessageAdvice;
+import org.springframework.lang.Nullable;
 import org.springframework.transaction.interceptor.DefaultTransactionAttribute;
 import org.springframework.transaction.interceptor.MatchAlwaysTransactionAttributeSource;
 import org.springframework.transaction.interceptor.TransactionInterceptor;
@@ -356,12 +357,14 @@ public abstract class IntegrationNamespaceUtils {
 	 * @param rootBuilder The root builder.
 	 * @param parserContext The parser context.
 	 * @param headerMapperBuilder The header mapper builder.
-	 * @param replyHeaderValue The reply header value.
+	 * @param replyHeaderValueArg The reply header value.
 	 */
 	public static void configureHeaderMapper(Element element, BeanDefinitionBuilder rootBuilder,
-			ParserContext parserContext, BeanDefinitionBuilder headerMapperBuilder, String replyHeaderValue) {
+			ParserContext parserContext, BeanDefinitionBuilder headerMapperBuilder,
+			@Nullable String replyHeaderValueArg) {
 
 		String defaultMappedReplyHeadersAttributeName = "mapped-reply-headers";
+		String replyHeaderValue = replyHeaderValueArg;
 		if (!StringUtils.hasText(replyHeaderValue)) {
 			replyHeaderValue = defaultMappedReplyHeadersAttributeName;
 		}

spring-integration-core/src/main/java/org/springframework/integration/dispatcher/AggregateMessageDeliveryException.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2018 the original author or authors.
+ * Copyright 2002-2019 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.
@@ -29,6 +29,7 @@ import org.springframework.util.StringUtils;
  *
  * @author Mark Fisher
  * @author Artem Bilan
+ * @author Gary Russell
  *
  * @since 1.0.3
  */
@@ -67,12 +68,14 @@ public class AggregateMessageDeliveryException extends MessageDeliveryException
 
 	private String appendPeriodIfNecessary(String baseMessage) {
 		if (!StringUtils.hasText(baseMessage)) {
-			baseMessage = "";
+			return "";
 		}
 		else if (!baseMessage.endsWith(".")) {
-			baseMessage = baseMessage + ".";
+			return baseMessage + ".";
+		}
+		else {
+			return baseMessage;
 		}
-		return baseMessage;
 	}
 
 }

spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2018 the original author or authors.
+ * Copyright 2016-2019 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.
@@ -221,13 +221,16 @@ public final class IntegrationFlows {
 	 * @see SourcePollingChannelAdapterSpec
 	 */
 	public static IntegrationFlowBuilder from(MessageSource<?> messageSource,
-			Consumer<SourcePollingChannelAdapterSpec> endpointConfigurer) {
+			@Nullable Consumer<SourcePollingChannelAdapterSpec> endpointConfigurer) {
+
 		return from(messageSource, endpointConfigurer, null);
 	}
 
 	private static IntegrationFlowBuilder from(MessageSource<?> messageSource,
-			Consumer<SourcePollingChannelAdapterSpec> endpointConfigurer,
-			IntegrationFlowBuilder integrationFlowBuilder) {
+			@Nullable Consumer<SourcePollingChannelAdapterSpec> endpointConfigurer,
+			@Nullable IntegrationFlowBuilder integrationFlowBuilderArg) {
+
+		IntegrationFlowBuilder integrationFlowBuilder = integrationFlowBuilderArg;
 		SourcePollingChannelAdapterSpec spec = new SourcePollingChannelAdapterSpec(messageSource);
 		if (endpointConfigurer != null) {
 			endpointConfigurer.accept(spec);
@@ -262,7 +265,9 @@ public final class IntegrationFlows {
 	}
 
 	private static IntegrationFlowBuilder from(MessageProducerSupport messageProducer,
-			IntegrationFlowBuilder integrationFlowBuilder) {
+			@Nullable IntegrationFlowBuilder integrationFlowBuilderArg) {
+
+		IntegrationFlowBuilder integrationFlowBuilder = integrationFlowBuilderArg;
 		MessageChannel outputChannel = messageProducer.getOutputChannel();
 		if (outputChannel == null) {
 			outputChannel = new DirectChannel();
@@ -354,8 +359,9 @@ public final class IntegrationFlows {
 	}
 
 	private static IntegrationFlowBuilder from(MessagingGatewaySupport inboundGateway,
-			IntegrationFlowBuilder integrationFlowBuilder) {
+			@Nullable IntegrationFlowBuilder integrationFlowBuilderArg) {
 
+		IntegrationFlowBuilder integrationFlowBuilder = integrationFlowBuilderArg;
 		MessageChannel outputChannel = inboundGateway.getRequestChannel();
 		if (outputChannel == null) {
 			outputChannel = new DirectChannel();

spring-integration-core/src/main/java/org/springframework/integration/dsl/context/StandardIntegrationFlowContext.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2018 the original author or authors.
+ * Copyright 2016-2019 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.
@@ -34,6 +34,7 @@ import org.springframework.beans.factory.support.BeanDefinitionRegistry;
 import org.springframework.integration.core.MessagingTemplate;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.support.context.NamedComponent;
+import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
@@ -124,7 +125,8 @@ public final class StandardIntegrationFlowContext implements IntegrationFlowCont
 	}
 
 	@SuppressWarnings("unchecked")
-	private Object registerBean(Object bean, String beanName, String parentName) {
+	private Object registerBean(Object bean, @Nullable String beanNameArg, String parentName) {
+		String beanName = beanNameArg;
 		if (beanName == null) {
 			beanName = generateBeanName(bean, parentName);
 		}

spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2018 the original author or authors.
+ * Copyright 2002-2019 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.
@@ -192,7 +192,8 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements
 	protected void doStop() {
 	}
 
-	protected void sendMessage(Message<?> message) {
+	protected void sendMessage(Message<?> messageArg) {
+		Message<?> message = messageArg;
 		if (message == null) {
 			throw new MessagingException("cannot send a null message");
 		}

spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2018 the original author or authors.
+ * Copyright 2002-2019 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.
@@ -228,7 +228,8 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint
 	}
 
 	@Override
-	protected void handleMessage(Message<?> message) {
+	protected void handleMessage(Message<?> messageArg) {
+		Message<?> message = messageArg;
 		if (this.shouldTrack) {
 			message = MessageHistory.write(message, this, getMessageBuilderFactory());
 		}

spring-integration-core/src/main/java/org/springframework/integration/expression/ReloadableResourceBundleExpressionSource.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2018 the original author or authors.
+ * Copyright 2002-2019 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.
@@ -408,9 +408,10 @@ public class ReloadableResourceBundleExpressionSource implements ExpressionSourc
 	 * The holder can be <code>null</code> if not cached before, or a timed-out cache entry
 	 * (potentially getting re-validated against the current last-modified timestamp).
 	 * @param filename the bundle filename (basename + Locale)
-	 * @param propHolder the current PropertiesHolder for the bundle
+	 * @param propHolderArg the current PropertiesHolder for the bundle
 	 */
-	private PropertiesHolder refreshProperties(String filename, PropertiesHolder propHolder) {
+	private PropertiesHolder refreshProperties(String filename, @Nullable PropertiesHolder propHolderArg) {
+		PropertiesHolder propHolder = propHolderArg;
 		long refreshTimestamp = (this.cacheMillis < 0) ? -1 : System.currentTimeMillis();
 
 		Resource resource = this.resourceLoader.getResource(filename + PROPERTIES_SUFFIX);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2018 the original author or authors.
+ * Copyright 2002-2019 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.
@@ -232,10 +232,9 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
 	 * from any object passed in a send or sendAndReceive operation.
 	 * @param requestMapper The request mapper.
 	 */
-	public void setRequestMapper(InboundMessageMapper<?> requestMapper) {
-		requestMapper = (requestMapper != null) ? requestMapper : new DefaultRequestMapper();
-		this.requestMapper = requestMapper;
-		this.messageConverter.setInboundMessageMapper(requestMapper);
+	public void setRequestMapper(@Nullable InboundMessageMapper<?> requestMapper) {
+		this.requestMapper = (requestMapper != null) ? requestMapper : new DefaultRequestMapper();
+		this.messageConverter.setInboundMessageMapper(this.requestMapper);
 	}
 
 	/**

spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2018 the original author or authors.
+ * Copyright 2002-2019 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.
@@ -139,7 +139,8 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport
 	}
 
 	@Override
-	public void handleMessage(Message<?> message) {
+	public void handleMessage(Message<?> messageArg) {
+		Message<?> message = messageArg;
 		Assert.notNull(message, "Message must not be null");
 		Assert.notNull(message.getPayload(), "Message payload must not be null"); //NOSONAR - false positive
 		if (this.loggingEnabled && this.logger.isDebugEnabled()) {

spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2018 the original author or authors.
+ * Copyright 2014-2019 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.
@@ -33,6 +33,7 @@ import org.springframework.integration.core.MessageProducer;
 import org.springframework.integration.core.MessagingTemplate;
 import org.springframework.integration.routingslip.RoutingSlipRouteStrategy;
 import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandlingException;
@@ -205,12 +206,8 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan
 	@Override
 	public MessageChannel getOutputChannel() {
 		if (this.outputChannelName != null) {
-			synchronized (this) {
-				if (this.outputChannelName != null) {
-					this.outputChannel = getChannelResolver().resolveDestination(this.outputChannelName);
-					this.outputChannelName = null;
-				}
-			}
+			this.outputChannel = getChannelResolver().resolveDestination(this.outputChannelName);
+			this.outputChannelName = null;
 		}
 		return this.outputChannel;
 	}
@@ -235,9 +232,10 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan
 		return false;
 	}
 
-	protected void produceOutput(Object reply, final Message<?> requestMessage) {
+	protected void produceOutput(Object replyArg, final Message<?> requestMessage) {
 		final MessageHeaders requestHeaders = requestMessage.getHeaders();
 
+		Object reply = replyArg;
 		Object replyChannel = null;
 		if (getOutputChannel() == null) {
 			Map<?, ?> routingSlipHeader = requestHeaders.get(IntegrationMessageHeaderAccessor.ROUTING_SLIP, Map.class);
@@ -252,20 +250,7 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan
 				AtomicInteger routingSlipIndex = new AtomicInteger((Integer) value);
 				replyChannel = getOutputChannelFromRoutingSlip(reply, requestMessage, routingSlip, routingSlipIndex);
 				if (replyChannel != null) {
-					//TODO Migrate to the SF MessageBuilder
-					AbstractIntegrationMessageBuilder<?> builder = null;
-					if (reply instanceof Message) {
-						builder = this.getMessageBuilderFactory().fromMessage((Message<?>) reply);
-					}
-					else if (reply instanceof AbstractIntegrationMessageBuilder) {
-						builder = (AbstractIntegrationMessageBuilder<?>) reply;
-					}
-					else {
-						builder = this.getMessageBuilderFactory().withPayload(reply);
-					}
-					builder.setHeader(IntegrationMessageHeaderAccessor.ROUTING_SLIP,
-							Collections.singletonMap(routingSlip, routingSlipIndex.get()));
-					reply = builder;
+					reply = addRoutingSlipHeader(reply, routingSlip, routingSlipIndex);
 				}
 			}
 
@@ -276,52 +261,16 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan
 				}
 			}
 		}
+		doProduceOutput(requestMessage, requestHeaders, reply, replyChannel);
+	}
+
+	private void doProduceOutput(final Message<?> requestMessage, final MessageHeaders requestHeaders, Object reply,
+			Object replyChannel) {
 
 		if (this.async && (reply instanceof ListenableFuture<?> || reply instanceof Publisher<?>)) {
 			if (reply instanceof ListenableFuture<?> ||
 					!(getOutputChannel() instanceof ReactiveStreamsSubscribableChannel)) {
-				ListenableFuture<?> future;
-				if (reply instanceof ListenableFuture<?>) {
-					future = (ListenableFuture<?>) reply;
-				}
-				else {
-					SettableListenableFuture<Object> settableListenableFuture = new SettableListenableFuture<>();
-
-					Mono.from((Publisher<?>) reply)
-							.subscribe(settableListenableFuture::set, settableListenableFuture::setException);
-
-					future = settableListenableFuture;
-				}
-
-				Object theReplyChannel = replyChannel;
-				future.addCallback(new ListenableFutureCallback<Object>() {
-
-					@Override
-					public void onSuccess(Object result) {
-						Message<?> replyMessage = null;
-						try {
-							replyMessage = createOutputMessage(result, requestHeaders);
-							sendOutput(replyMessage, theReplyChannel, false);
-						}
-						catch (Exception e) {
-							Exception exceptionToLogAndSend = e;
-							if (!(e instanceof MessagingException)) {
-								exceptionToLogAndSend = new MessageHandlingException(requestMessage, e);
-								if (replyMessage != null) {
-									exceptionToLogAndSend = new MessagingException(replyMessage, exceptionToLogAndSend);
-								}
-							}
-							logger.error("Failed to send async reply: " + result.toString(), exceptionToLogAndSend);
-							onFailure(exceptionToLogAndSend);
-						}
-					}
-
-					@Override
-					public void onFailure(Throwable ex) {
-						sendErrorMessage(requestMessage, ex);
-					}
-
-				});
+				asyncNonReactiveReply(requestMessage, requestHeaders, reply, replyChannel);
 			}
 			else {
 				((ReactiveStreamsSubscribableChannel) getOutputChannel())
@@ -335,6 +284,71 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan
 		}
 	}
 
+	private AbstractIntegrationMessageBuilder<?> addRoutingSlipHeader(Object reply, List<?> routingSlip,
+			AtomicInteger routingSlipIndex) {
+
+		//TODO Migrate to the SF MessageBuilder
+		AbstractIntegrationMessageBuilder<?> builder = null;
+		if (reply instanceof Message) {
+			builder = this.getMessageBuilderFactory().fromMessage((Message<?>) reply);
+		}
+		else if (reply instanceof AbstractIntegrationMessageBuilder) {
+			builder = (AbstractIntegrationMessageBuilder<?>) reply;
+		}
+		else {
+			builder = this.getMessageBuilderFactory().withPayload(reply);
+		}
+		builder.setHeader(IntegrationMessageHeaderAccessor.ROUTING_SLIP,
+				Collections.singletonMap(routingSlip, routingSlipIndex.get()));
+		return builder;
+	}
+
+	private void asyncNonReactiveReply(final Message<?> requestMessage, final MessageHeaders requestHeaders,
+			Object reply, Object replyChannel) {
+		ListenableFuture<?> future;
+		if (reply instanceof ListenableFuture<?>) {
+			future = (ListenableFuture<?>) reply;
+		}
+		else {
+			SettableListenableFuture<Object> settableListenableFuture = new SettableListenableFuture<>();
+
+			Mono.from((Publisher<?>) reply)
+					.subscribe(settableListenableFuture::set, settableListenableFuture::setException);
+
+			future = settableListenableFuture;
+		}
+
+		Object theReplyChannel = replyChannel;
+		future.addCallback(new ListenableFutureCallback<Object>() {
+
+			@Override
+			public void onSuccess(Object result) {
+				Message<?> replyMessage = null;
+				try {
+					replyMessage = createOutputMessage(result, requestHeaders);
+					sendOutput(replyMessage, theReplyChannel, false);
+				}
+				catch (Exception e) {
+					Exception exceptionToLogAndSend = e;
+					if (!(e instanceof MessagingException)) {
+						exceptionToLogAndSend = new MessageHandlingException(requestMessage, e);
+						if (replyMessage != null) {
+							exceptionToLogAndSend = new MessagingException(replyMessage, exceptionToLogAndSend);
+						}
+					}
+					logger.error("Failed to send async reply: " + result.toString(), exceptionToLogAndSend);
+					onFailure(exceptionToLogAndSend);
+				}
+			}
+
+			@Override
+			public void onFailure(Throwable ex) {
+				sendErrorMessage(requestMessage, ex);
+			}
+
+		});
+	}
+
 	private Object getOutputChannelFromRoutingSlip(Object reply, Message<?> requestMessage, List<?> routingSlip,
 			AtomicInteger routingSlipIndex) {
 		if (routingSlipIndex.get() >= routingSlip.size()) {
@@ -397,11 +411,12 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan
 	 * 'outputChannel' is <code>null</code>. In that case, the 'replyChannel' value must not also be
 	 * <code>null</code>, and it must be an instance of either String or {@link MessageChannel}.
 	 * @param output the output object to send
-	 * @param replyChannel the 'replyChannel' value from the original request
+	 * @param replyChannelArg the 'replyChannel' value from the original request
 	 * @param useArgChannel - use the replyChannel argument (must not be null), not
 	 * the configured output channel.
 	 */
-	protected void sendOutput(Object output, Object replyChannel, boolean useArgChannel) {
+	protected void sendOutput(Object output, @Nullable Object replyChannelArg, boolean useArgChannel) {
+		Object replyChannel = replyChannelArg;
 		MessageChannel outputChannel = getOutputChannel();
 		if (!useArgChannel && outputChannel != null) {
 			replyChannel = outputChannel;

spring-integration-core/src/main/java/org/springframework/integration/history/MessageHistory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2019 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.
@@ -44,6 +44,7 @@ import org.springframework.util.StringUtils;
 /**
  * @author Mark Fisher
  * @author Artem Bilan
+ * @author Gary Russell
  * @since 2.0
  */
 @SuppressWarnings("serial")
@@ -74,8 +75,10 @@ public final class MessageHistory implements List<Properties>, Serializable {
 	}
 
 	@SuppressWarnings("unchecked")
-	public static <T> Message<T> write(Message<T> message, NamedComponent component,
+	public static <T> Message<T> write(Message<T> messageArg, NamedComponent component,
 			MessageBuilderFactory messageBuilderFactory) {
+
+		Message<T> message = messageArg;
 		Assert.notNull(message, "Message must not be null");
 		Assert.notNull(component, "Component must not be null");
 		Properties metadata = extractMetadata(component);

spring-integration-core/src/main/java/org/springframework/integration/transformer/ObjectToMapTransformer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 the original author or authors.
+ * Copyright 2002-2019 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.
@@ -129,7 +129,8 @@ public class ObjectToMapTransformer extends AbstractPayloadTransformer<Object, M
 		return resultMap;
 	}
 
-	private void doFlatten(String propertyPrefix, Map<String, Object> inputMap, Map<String, Object> resultMap) {
+	private void doFlatten(String propertyPrefixArg, Map<String, Object> inputMap, Map<String, Object> resultMap) {
+		String propertyPrefix = propertyPrefixArg;
 		if (StringUtils.hasText(propertyPrefix)) {
 			propertyPrefix = propertyPrefix + ".";
 		}