GH-2770: AbstractMessageSource EvaluationContext

Fixes #2770

Only create an evaluation context for header expressions if we actually
have header expressions.

* Some `AbstractMessageSource` refactoring to avoid unnecessary code and
some optimization

**Cherry-pick to 5.1.x**

Author: garyrussell

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

@@ -16,7 +16,7 @@
 
 package org.springframework.integration.endpoint;
 
-import java.util.Collections;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicLong;
 
@@ -33,7 +33,6 @@
 import org.springframework.integration.util.AbstractExpressionEvaluator;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
-import org.springframework.messaging.MessagingException;
 import org.springframework.util.CollectionUtils;
 
 /**
@@ -52,7 +51,7 @@ public abstract class AbstractMessageSource<T> extends AbstractExpressionEvaluat
 
 	private final ManagementOverrides managementOverrides = new ManagementOverrides();
 
-	private volatile Map<String, Expression> headerExpressions = Collections.emptyMap();
+	private Map<String, Expression> headerExpressions;
 
 	private String beanName;
 
@@ -68,9 +67,10 @@ public abstract class AbstractMessageSource<T> extends AbstractExpressionEvaluat
 
 	private CounterFacade receiveCounter;
 
-	public void setHeaderExpressions(Map<String, Expression> headerExpressions) {
-		this.headerExpressions = (headerExpressions != null)
-				? headerExpressions : Collections.emptyMap();
+	public void setHeaderExpressions(@Nullable Map<String, Expression> headerExpressions) {
+		if (!CollectionUtils.isEmpty(headerExpressions)) {
+			this.headerExpressions = new HashMap<>(headerExpressions);
+		}
 	}
 
 	@Override
@@ -162,68 +162,61 @@ public final Message<T> receive() {
 
 	@SuppressWarnings("unchecked")
 	protected Message<T> buildMessage(Object result) {
-		Message<T> message = null;
+		Message<?> message = null;
 		Map<String, Object> headers = evaluateHeaders();
-		if (result instanceof AbstractIntegrationMessageBuilder) {
+		if (result instanceof AbstractIntegrationMessageBuilder<?>) {
 			if (!CollectionUtils.isEmpty(headers)) {
-				((AbstractIntegrationMessageBuilder<T>) result).copyHeaders(headers);
+				((AbstractIntegrationMessageBuilder<?>) result).copyHeaders(headers);
 			}
-			message = ((AbstractIntegrationMessageBuilder<T>) result).build();
+			message = ((AbstractIntegrationMessageBuilder<?>) result).build();
 		}
 		else if (result instanceof Message<?>) {
-			try {
-				message = (Message<T>) result;
-			}
-			catch (Exception e) {
-				throw new MessagingException("MessageSource returned unexpected type.", e);
-			}
+			message = (Message<?>) result;
 			if (!CollectionUtils.isEmpty(headers)) {
 				// create a new Message from this one in order to apply headers
-				message = getMessageBuilderFactory()
-						.fromMessage(message)
-						.copyHeaders(headers)
-						.build();
+				message =
+						getMessageBuilderFactory()
+								.fromMessage(message)
+								.copyHeaders(headers)
+								.build();
 			}
 		}
 		else if (result != null) {
-			T payload;
-			try {
-				payload = (T) result;
-			}
-			catch (Exception e) {
-				throw new MessagingException("MessageSource returned unexpected type.", e);
-			}
-			message = getMessageBuilderFactory()
-					.withPayload(payload)
-					.copyHeaders(headers)
-					.build();
+			message =
+					getMessageBuilderFactory()
+							.withPayload(result)
+							.copyHeaders(headers)
+							.build();
 		}
 		if (this.countsEnabled && message != null) {
 			if (this.metricsCaptor != null) {
 				incrementReceiveCounter();
 			}
 			this.messageCount.incrementAndGet();
 		}
-		return message;
+		return (Message<T>) message;
 	}
 
 	private void incrementReceiveCounter() {
 		if (this.receiveCounter == null) {
 			this.receiveCounter = this.metricsCaptor.counterBuilder(RECEIVE_COUNTER_NAME)
-				.tag("name", getComponentName() == null ? "unknown" : getComponentName())
-				.tag("type", "source")
-				.tag("result", "success")
-				.tag("exception", "none")
-				.description("Messages received")
-				.build();
+					.tag("name", getComponentName() == null ? "unknown" : getComponentName())
+					.tag("type", "source")
+					.tag("result", "success")
+					.tag("exception", "none")
+					.description("Messages received")
+					.build();
 		}
 		this.receiveCounter.increment();
 	}
 
+	@Nullable
 	private Map<String, Object> evaluateHeaders() {
-		return ExpressionEvalMap.from(this.headerExpressions)
-				.usingEvaluationContext(getEvaluationContext())
-				.build();
+		return CollectionUtils.isEmpty(this.headerExpressions)
+				? null
+				: ExpressionEvalMap.from(this.headerExpressions)
+						.usingEvaluationContext(getEvaluationContext())
+						.build();
 	}
 
 	/**