Fix ambiguity for NullChannel instances bean name

When we use explicitly `new NullChannel()` in DSL definition, we don't
provide a bean name and just end up with an explicit `nullChannel`
component name in the class.
This way it clashes with the global `errorChannel`, but since instances
are different we get a `BeanDefinitionOverrideException`

* Remove explicit `nullChannel` component name in the `NullChannel` in
favor of the generated name as it is done for any regular beans
* Some internal polishing in the `NullChannel`
* Some polishing in the `IntegrationFlowBeanPostProcessor` to deal with
`NullChannel` as with any other regular `AbstractMessageChannel`s and
generate a proper name for it within the scope of the current
`IntegrationFlow`

spring-integration-core/src/main/java/org/springframework/integration/channel/NullChannel.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.
@@ -31,10 +31,10 @@ import org.springframework.integration.support.management.MessageChannelMetrics;
 import org.springframework.integration.support.management.Statistics;
 import org.springframework.integration.support.management.metrics.MetricsCaptor;
 import org.springframework.integration.support.management.metrics.TimerFacade;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.util.Assert;
-import org.springframework.util.StringUtils;
 
 /**
  * A channel implementation that essentially behaves like "/dev/null".
@@ -54,13 +54,13 @@ public class NullChannel implements PollableChannel, MessageChannelMetrics,
 
 	private final ManagementOverrides managementOverrides = new ManagementOverrides();
 
-	private volatile AbstractMessageChannelMetrics channelMetrics = new DefaultMessageChannelMetrics("nullChannel");
+	private AbstractMessageChannelMetrics channelMetrics = new DefaultMessageChannelMetrics("nullChannel");
 
-	private volatile boolean countsEnabled;
+	private boolean countsEnabled;
 
-	private volatile boolean statsEnabled;
+	private boolean statsEnabled;
 
-	private volatile boolean loggingEnabled = true;
+	private boolean loggingEnabled = true;
 
 	private String beanName;
 
@@ -71,7 +71,7 @@ public class NullChannel implements PollableChannel, MessageChannelMetrics,
 	@Override
 	public void setBeanName(String beanName) {
 		this.beanName = beanName;
-		this.channelMetrics = new DefaultMessageChannelMetrics(getComponentName());
+		this.channelMetrics = new DefaultMessageChannelMetrics(this.beanName);
 	}
 
 	@Override
@@ -86,13 +86,14 @@ public class NullChannel implements PollableChannel, MessageChannelMetrics,
 	}
 
 	@Override
+	@Nullable
 	public String getComponentName() {
-		return StringUtils.hasText(this.beanName) ? this.beanName : "nullChannel";
+		return this.beanName;
 	}
 
 	@Override
 	public String getComponentType() {
-		return "channel";
+		return "null-channel";
 	}
 
 	@Override
@@ -246,7 +247,7 @@ public class NullChannel implements PollableChannel, MessageChannelMetrics,
 		if (this.successTimer == null) {
 			this.successTimer = this.metricsCaptor.timerBuilder(SEND_TIMER_NAME)
 					.tag("type", "channel")
-					.tag("name", getComponentName() == null ? "unknown" : getComponentName())
+					.tag("name", getComponentName() == null ? "nullChannel" : getComponentName())
 					.tag("result", "success")
 					.tag("exception", "none")
 					.description("Subflow process time")
@@ -257,7 +258,7 @@ public class NullChannel implements PollableChannel, MessageChannelMetrics,
 
 	@Override
 	public Message<?> receive() {
-		if (this.loggingEnabled && this.logger.isDebugEnabled()) {
+		if (this.loggingEnabled) {
 			this.logger.debug("receive called on null channel");
 		}
 		return null;
@@ -265,7 +266,7 @@ public class NullChannel implements PollableChannel, MessageChannelMetrics,
 
 	@Override
 	public Message<?> receive(long timeout) {
-		return this.receive();
+		return receive();
 	}
 
 	@Override

spring-integration-core/src/main/java/org/springframework/integration/dsl/context/IntegrationFlowBeanPostProcessor.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.
@@ -52,6 +52,7 @@ import org.springframework.core.io.DescriptiveResource;
 import org.springframework.integration.channel.AbstractMessageChannel;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.FixedSubscriberChannel;
+import org.springframework.integration.channel.NullChannel;
 import org.springframework.integration.config.ConsumerEndpointFactoryBean;
 import org.springframework.integration.config.IntegrationConfigUtils;
 import org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean;
@@ -175,8 +176,8 @@ public class IntegrationFlowBeanPostProcessor
 			}
 			else {
 				if (noBeanPresentForComponent(component, flowBeanName)) {
-					if (component instanceof AbstractMessageChannel) {
-						String channelBeanName = ((AbstractMessageChannel) component).getComponentName();
+					if (component instanceof AbstractMessageChannel || component instanceof NullChannel) {
+						String channelBeanName = ((NamedComponent) component).getComponentName();
 						if (channelBeanName == null) {
 							channelBeanName = entry.getValue();
 							if (channelBeanName == null) {
@@ -263,7 +264,7 @@ public class IntegrationFlowBeanPostProcessor
 						registerComponent(gateway, gatewayId, flowBeanName,
 								beanDefinition -> {
 									((AbstractBeanDefinition) beanDefinition)
-											.setSource(new DescriptiveResource(gateway.getObjectType().getName()));
+											.setSource(new DescriptiveResource("" + gateway.getObjectType()));
 								});
 
 						targetIntegrationComponents.put(component, gatewayId);

spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.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.
@@ -504,6 +504,26 @@ public class IntegrationFlowTests {
 		this.nullChannel.setCountsEnabled(false);
 	}
 
+	@Autowired
+	@Qualifier("flowWithLocalNullChannel.input")
+	private MessageChannel flowWithLocalNullChannelInput;
+
+	@Autowired
+	@Qualifier("flowWithLocalNullChannel.channel#0")
+	private NullChannel localNullChannel;
+
+	@Test
+	public void testLocalNullChannel() {
+		this.localNullChannel.setCountsEnabled(true);
+
+		this.flowWithLocalNullChannelInput.send(new GenericMessage<>("foo"));
+
+		assertEquals(1, this.localNullChannel.getSendCount());
+
+		assertNotSame(this.nullChannel, this.localNullChannel);
+	}
+
+
 	@Autowired
 	private EventDrivenConsumer flow1WithPrototypeHandlerConsumer;
 
@@ -892,6 +912,11 @@ public class IntegrationFlowTests {
 					.nullChannel();
 		}
 
+		@Bean
+		public IntegrationFlow flowWithLocalNullChannel() {
+			return f -> f.channel(new NullChannel());
+		}
+
 		@Bean
 		@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
 		public AbstractReplyProducingMessageHandler myHandler() {