Fix DSL to deal with beanNames for handlers (#2707)

* Fix DSL to deal with beanNames for handlers

When consumer endpoint is created by the Framework, the target
`MessageHandler` gets a `componentName` from the `ConsumerEndpointFactoryBean`.
Therefore we can't rely on the `getComponentName()` when we create beans
from Java DSL.

* Introduce `NamedComponent.getBeanName()` contract; make it default
to the `getComponentName()`; implement this method from the
`IntegrationObjectSupport`
* Use this new `getBeanName()` from the `IntegrationFlowBeanPostProcessor`
and `StandardIntegrationFlowContext` for better existing beans checks
and possible reuse existing `MessageHandler` in different endpoints
* Optimize `FixedSubscriberChannel` and implement `getBeanName()` over there
* Implement `getBeanName()` in the `AbstractMessageSource`; use
`ExpressionEvalMap` to evaluate headers expressions
* Implement `getBeanName()` in `Trackable*Metrics` classes as delegation
to the `this.trackable`

* * Fix race condition around `discardChannelName` property in the `MessageFilter`

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2016 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.
@@ -34,16 +34,18 @@ import org.springframework.messaging.SubscribableChannel;
  * <b>Note: Stopping ({@link #unsubscribe(MessageHandler)}) the subscribed ({@link MessageHandler}) has no effect.</b>
  *
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 4.0
  *
  */
 public final class FixedSubscriberChannel implements SubscribableChannel, BeanNameAware, NamedComponent {
 
-	private final Log logger = LogFactory.getLog(FixedSubscriberChannel.class);
+	private static final Log LOGGER = LogFactory.getLog(FixedSubscriberChannel.class);
 
 	private final MessageHandler handler;
 
-	private volatile String beanName;
+	private String beanName;
 
 	public FixedSubscriberChannel() {
 		throw new IllegalArgumentException("Cannot instantiate a " + this.getClass().getSimpleName()
@@ -59,9 +61,14 @@ public final class FixedSubscriberChannel implements SubscribableChannel, BeanNa
 		this.beanName = name;
 	}
 
+	@Override
+	public String getBeanName() {
+		return this.beanName;
+	}
+
 	@Override
 	public boolean send(Message<?> message) {
-		return this.send(message, 0);
+		return send(message, 0);
 	}
 
 	@Override
@@ -83,23 +90,23 @@ public final class FixedSubscriberChannel implements SubscribableChannel, BeanNa
 
 	@Override
 	public boolean subscribe(MessageHandler handler) {
-		if (handler != this.handler && this.logger.isDebugEnabled()) {
-			this.logger.debug(this.getComponentName() + ": cannot be subscribed to (it has a fixed single subscriber).");
+		if (handler != this.handler && LOGGER.isDebugEnabled()) {
+			LOGGER.debug(getComponentName() + ": cannot be subscribed to (it has a fixed single subscriber).");
 		}
 		return false;
 	}
 
 	@Override
 	public boolean unsubscribe(MessageHandler handler) {
-		if (this.logger.isDebugEnabled()) {
-			this.logger.debug(this.getComponentName() + ": cannot be unsubscribed from (it has a fixed single subscriber).");
+		if (LOGGER.isDebugEnabled()) {
+			LOGGER.debug(getComponentName() + ": cannot be unsubscribed from (it has a fixed single subscriber).");
 		}
 		return false;
 	}
 
 	@Override
 	public String getComponentType() {
-		return "Fixed Subscriber Channel";
+		return "fixed-subscriber-channel";
 	}
 
 	@Override

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

@@ -85,6 +85,12 @@ public class NullChannel implements PollableChannel, MessageChannelMetrics,
 		this.managementOverrides.loggingConfigured = true;
 	}
 
+	@Override
+	@Nullable
+	public String getBeanName() {
+		return this.beanName;
+	}
+
 	@Override
 	@Nullable
 	public String getComponentName() {

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

@@ -219,7 +219,7 @@ public abstract class AbstractSimpleMessageHandlerFactoryBean<H extends MessageH
 				else if (this.logger.isDebugEnabled()) {
 					String name = this.componentName;
 					if (name == null && actualHandler instanceof NamedComponent) {
-						name = ((NamedComponent) actualHandler).getComponentName();
+						name = ((NamedComponent) actualHandler).getBeanName();
 					}
 					this.logger.debug("adviceChain can only be set on an AbstractReplyProducingMessageHandler"
 							+ (name == null ? "" : (", " + name)) + ".");

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

@@ -167,7 +167,7 @@ public abstract class AbstractStandardMessageHandlerFactoryBean
 	private void checkReuse(AbstractMessageProducingHandler replyHandler) {
 		Assert.isTrue(!referencedReplyProducers.contains(replyHandler),
 				"An AbstractMessageProducingMessageHandler may only be referenced once (" +
-						replyHandler.getComponentName() + ") - use scope=\"prototype\"");
+						replyHandler.getBeanName() + ") - use scope=\"prototype\"");
 		referencedReplyProducers.add(replyHandler);
 		this.replyHandler = replyHandler;
 	}
@@ -213,7 +213,7 @@ public abstract class AbstractStandardMessageHandlerFactoryBean
 			else {
 				if (this.requiresReply && logger.isDebugEnabled()) {
 					logger.debug("requires-reply can only be set to AbstractReplyProducingMessageHandler " +
-							"or its subclass, " + handler.getComponentName() + " doesn't support it.");
+							"or its subclass, " + handler.getBeanName() + " doesn't support it.");
 				}
 			}
 		}

spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.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.
@@ -70,7 +70,7 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo
 
 	protected static final ExpressionParser EXPRESSION_PARSER = new SpelExpressionParser();
 
-	private static final IdGenerator idGenerator = new AlternativeJdkIdGenerator(); // NOSONAR lower case
+	private static final IdGenerator ID_GENERATOR = new AlternativeJdkIdGenerator();
 
 	/**
 	 * Logger that is available to subclasses
@@ -106,6 +106,11 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo
 		this.beanName = beanName;
 	}
 
+	@Override
+	public String getBeanName() {
+		return this.beanName;
+	}
+
 	/**
 	 * Will return the name of this component identified by {@link #componentName} field.
 	 * If {@link #componentName} was not set this method will default to the 'beanName' of this component;
@@ -224,7 +229,7 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo
 			this.conversionService = IntegrationUtils.getConversionService(this.beanFactory);
 			if (this.conversionService == null && this.logger.isDebugEnabled()) {
 				this.logger.debug("Unable to attempt conversion of Message payload types. Component '" +
-						this.getComponentName() + "' has no explicit ConversionService reference, " +
+						getComponentName() + "' has no explicit ConversionService reference, " +
 						"and there is no 'integrationConversionService' bean within the context.");
 			}
 		}
@@ -301,7 +306,7 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo
 	}
 
 	public static UUID generateId() {
-		return idGenerator.generateId();
+		return ID_GENERATOR.generateId();
 	}
 
 }

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

@@ -389,7 +389,7 @@ public class IntegrationFlowBeanPostProcessor
 	@SuppressWarnings("unchecked")
 	private boolean noBeanPresentForComponent(Object instance, String parentBeanName) {
 		if (instance instanceof NamedComponent) {
-			String beanName = ((NamedComponent) instance).getComponentName();
+			String beanName = ((NamedComponent) instance).getBeanName();
 			if (beanName != null) {
 				if (this.beanFactory.containsBean(beanName)) {
 					BeanDefinition existingBeanDefinition = this.beanFactory.getBeanDefinition(beanName);
@@ -445,10 +445,11 @@ public class IntegrationFlowBeanPostProcessor
 	}
 
 	private String generateBeanName(Object instance, String prefix, String fallbackId, boolean useFlowIdAsPrefix) {
-		if (instance instanceof NamedComponent && ((NamedComponent) instance).getComponentName() != null) {
+		if (instance instanceof NamedComponent && ((NamedComponent) instance).getBeanName() != null) {
+			String beanName = ((NamedComponent) instance).getBeanName();
 			return useFlowIdAsPrefix
-					? prefix + ((NamedComponent) instance).getComponentName()
-					: ((NamedComponent) instance).getComponentName();
+					? prefix + beanName
+					: beanName;
 		}
 		else if (fallbackId != null) {
 			return useFlowIdAsPrefix

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

@@ -216,9 +216,13 @@ public final class StandardIntegrationFlowContext implements IntegrationFlowCont
 	}
 
 	private String generateBeanName(Object instance, String parentName) {
-		if (instance instanceof NamedComponent && ((NamedComponent) instance).getComponentName() != null) {
-			return ((NamedComponent) instance).getComponentName();
+		if (instance instanceof NamedComponent) {
+			String beanName = ((NamedComponent) instance).getBeanName();
+			if (beanName != null) {
+				return beanName;
+			}
 		}
+
 		String generatedBeanName = (parentName != null ? parentName : "") + instance.getClass().getName();
 		String id = generatedBeanName;
 		int counter = -1;

spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractMessageSource.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.
@@ -17,13 +17,13 @@
 package org.springframework.integration.endpoint;
 
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicLong;
 
 import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.expression.Expression;
 import org.springframework.integration.core.MessageSource;
+import org.springframework.integration.expression.ExpressionEvalMap;
 import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
 import org.springframework.integration.support.context.NamedComponent;
 import org.springframework.integration.support.management.IntegrationManagedResource;
@@ -60,9 +60,9 @@ public abstract class AbstractMessageSource<T> extends AbstractExpressionEvaluat
 
 	private String managedName;
 
-	private volatile boolean countsEnabled;
+	private boolean countsEnabled;
 
-	private volatile boolean loggingEnabled = true;
+	private boolean loggingEnabled = true;
 
 	private MetricsCaptor metricsCaptor;
 
@@ -83,6 +83,11 @@ public abstract class AbstractMessageSource<T> extends AbstractExpressionEvaluat
 		this.beanName = name;
 	}
 
+	@Override
+	public String getBeanName() {
+		return this.beanName;
+	}
+
 	@Override
 	public void setManagedType(String managedType) {
 		this.managedType = managedType;
@@ -216,14 +221,9 @@ public abstract class AbstractMessageSource<T> extends AbstractExpressionEvaluat
 	}
 
 	private Map<String, Object> evaluateHeaders() {
-		Map<String, Object> results = new HashMap<>();
-		for (Map.Entry<String, Expression> entry : this.headerExpressions.entrySet()) {
-			Object headerValue = this.evaluateExpression(entry.getValue());
-			if (headerValue != null) {
-				results.put(entry.getKey(), headerValue);
-			}
-		}
-		return results;
+		return ExpressionEvalMap.from(this.headerExpressions)
+				.usingEvaluationContext(getEvaluationContext())
+				.build();
 	}
 
 	/**

spring-integration-core/src/main/java/org/springframework/integration/filter/MessageFilter.java

@@ -16,6 +16,7 @@
 
 package org.springframework.integration.filter;
 
+import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
 import org.springframework.context.Lifecycle;
 import org.springframework.core.convert.ConversionService;
@@ -47,11 +48,11 @@ public class MessageFilter extends AbstractReplyProducingPostProcessingMessageHa
 
 	private final MessageSelector selector;
 
-	private volatile boolean throwExceptionOnRejection;
+	private boolean throwExceptionOnRejection;
 
-	private volatile MessageChannel discardChannel;
+	private MessageChannel discardChannel;
 
-	private volatile String discardChannelName;
+	private String discardChannelName;
 
 	/**
 	 * Create a MessageFilter that will delegate to the given {@link MessageSelector}.
@@ -109,13 +110,10 @@ public class MessageFilter extends AbstractReplyProducingPostProcessingMessageHa
 
 	@Override
 	public MessageChannel getDiscardChannel() {
-		if (this.discardChannelName != null) {
-			synchronized (this) {
-				if (this.discardChannelName != null) {
-					this.discardChannel = getChannelResolver().resolveDestination(this.discardChannelName);
-					this.discardChannelName = null;
-				}
-			}
+		String channelName = this.discardChannelName;
+		if (channelName != null) {
+			this.discardChannel = getChannelResolver().resolveDestination(channelName);
+			this.discardChannelName = null;
 		}
 		return this.discardChannel;
 	}
@@ -129,15 +127,16 @@ public class MessageFilter extends AbstractReplyProducingPostProcessingMessageHa
 	@Override
 	protected void doInit() {
 		Assert.state(!(this.discardChannelName != null && this.discardChannel != null),
-					"'discardChannelName' and 'discardChannel' are mutually exclusive.");
+				"'discardChannelName' and 'discardChannel' are mutually exclusive.");
 		if (this.selector instanceof AbstractMessageProcessingSelector) {
 			ConversionService conversionService = getConversionService();
 			if (conversionService != null) {
 				((AbstractMessageProcessingSelector) this.selector).setConversionService(conversionService);
 			}
 		}
-		if (this.selector instanceof BeanFactoryAware && this.getBeanFactory() != null) {
-			((BeanFactoryAware) this.selector).setBeanFactory(this.getBeanFactory());
+		BeanFactory beanFactory = getBeanFactory();
+		if (this.selector instanceof BeanFactoryAware && beanFactory != null) {
+			((BeanFactoryAware) this.selector).setBeanFactory(beanFactory);
 		}
 	}
 
@@ -178,8 +177,7 @@ public class MessageFilter extends AbstractReplyProducingPostProcessingMessageHa
 				this.messagingTemplate.send(channel, message);
 			}
 			if (this.throwExceptionOnRejection) {
-				throw new MessageRejectedException(message, "MessageFilter '" + this.getComponentName()
-						+ "' rejected Message");
+				throw new MessageRejectedException(message, "MessageFilter '" + getBeanName() + "' rejected Message");
 			}
 		}
 		return result;

spring-integration-core/src/main/java/org/springframework/integration/support/context/NamedComponent.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 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.
@@ -18,6 +18,8 @@ package org.springframework.integration.support.context;
 
 /**
  * @author Mark Fisher
+ * @author Artem Bilan
+ *
  * @since 2.0
  */
 public interface NamedComponent {
@@ -26,4 +28,8 @@ public interface NamedComponent {
 
 	String getComponentType();
 
+	default String getBeanName() {
+		return getComponentName();
+	}
+
 }

spring-integration-core/src/main/java/org/springframework/integration/support/management/LifecycleTrackableMessageHandlerMetrics.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 the original author or authors.
+ * Copyright 2015-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.
@@ -23,6 +23,8 @@ import org.springframework.util.Assert;
  * Adds {@link TrackableComponent}.
  *
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 4.2
  */
 @IntegrationManagedResource
@@ -37,6 +39,11 @@ public class LifecycleTrackableMessageHandlerMetrics extends LifecycleMessageHan
 		this.trackable = (TrackableComponent) delegate;
 	}
 
+	@Override
+	public String getBeanName() {
+		return this.trackable.getBeanName();
+	}
+
 	@Override
 	public String getComponentName() {
 		return this.trackable.getComponentName();

spring-integration-core/src/main/java/org/springframework/integration/support/management/LifecycleTrackableMessageSourceMetrics.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 the original author or authors.
+ * Copyright 2015-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.
@@ -23,6 +23,8 @@ import org.springframework.util.Assert;
  * Adds {@link TrackableComponent}.
  *
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 2.0
  */
 @IntegrationManagedResource
@@ -37,6 +39,11 @@ public class LifecycleTrackableMessageSourceMetrics extends LifecycleMessageSour
 		this.trackable = (TrackableComponent) lifecycle;
 	}
 
+	@Override
+	public String getBeanName() {
+		return this.trackable.getBeanName();
+	}
+
 	@Override
 	public String getComponentName() {
 		return this.trackable.getComponentName();

spring-integration-core/src/main/java/org/springframework/integration/support/management/TrackableRouterMetrics.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 the original author or authors.
+ * Copyright 2015-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.
@@ -23,6 +23,8 @@ import org.springframework.util.Assert;
  * Adds {@link TrackableComponent}.
  *
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 2.0
  */
 public class TrackableRouterMetrics extends RouterMetrics implements TrackableComponent {
@@ -35,6 +37,11 @@ public class TrackableRouterMetrics extends RouterMetrics implements TrackableCo
 		this.trackable = (TrackableComponent) delegate;
 	}
 
+	@Override
+	public String getBeanName() {
+		return this.trackable.getBeanName();
+	}
+
 	@Override
 	public String getComponentName() {
 		return this.trackable.getComponentName();

spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java

@@ -72,6 +72,7 @@ import org.springframework.integration.dsl.Transformers;
 import org.springframework.integration.endpoint.EventDrivenConsumer;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.integration.handler.GenericHandler;
+import org.springframework.integration.handler.LoggingHandler;
 import org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer;
 import org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice;
 import org.springframework.integration.handler.advice.RequestHandlerRetryAdvice;
@@ -697,11 +698,16 @@ public class IntegrationFlowTests {
 			return tpte;
 		}
 
+		@Bean
+		public MessageHandler loggingMessageHandler() {
+			return new LoggingHandler(LoggingHandler.Level.DEBUG);
+		}
+
 		@Bean
 		public IntegrationFlow wireTapFlow1() {
 			return IntegrationFlows.from("tappedChannel1")
 					.wireTap("tapChannel", wt -> wt.selector(m -> m.getPayload().equals("foo")))
-					.channel("nullChannel")
+					.handle(loggingMessageHandler())
 					.get();
 		}
 
@@ -709,7 +715,7 @@ public class IntegrationFlowTests {
 		public IntegrationFlow wireTapFlow2() {
 			return f -> f
 					.wireTap("tapChannel", wt -> wt.selector(m -> m.getPayload().equals("foo")))
-					.channel("nullChannel");
+					.handle(loggingMessageHandler());
 		}
 
 		@Bean
@@ -717,7 +723,7 @@ public class IntegrationFlowTests {
 			return f -> f
 					.transform("payload")
 					.wireTap("tapChannel", wt -> wt.selector("payload == 'foo'"))
-					.channel("nullChannel");
+					.handle(loggingMessageHandler());
 		}
 
 		@Bean