INT-4572: Add MessageProducer.setOutputChannelName

JIRA: https://jira.spring.io/browse/INT-4572

* Add `setOutputChannelName()` contract into the `MessageProducer`
to avoid proxy unwrapping and casting to the `MessageProducerSupport`

spring-integration-core/src/main/java/org/springframework/integration/core/MessageProducer.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.
@@ -20,20 +20,30 @@ import org.springframework.messaging.MessageChannel;
 
 /**
  * Base interface for any component that is capable of sending
- * Messages to a {@link MessageChannel}.
+ * messages to a {@link MessageChannel}.
  *
  * @author Mark Fisher
+ * @author Artem Bilan
+ *
  * @since 2.0
  */
 public interface MessageProducer {
 
 	/**
-	 * Specify the MessageChannel to which produced Messages should be sent.
-	 *
+	 * Specify the {@link MessageChannel} to which produced Messages should be sent.
 	 * @param outputChannel The output channel.
 	 */
 	void setOutputChannel(MessageChannel outputChannel);
 
+	/**
+	 * Specify the bean name of the {@link MessageChannel} to which produced Messages should be sent.
+	 * @param outputChannel The output channel bean name.
+	 * @since 5.1.2
+	 */
+	default void setOutputChannelName(String outputChannel) {
+		throw new UnsupportedOperationException("This MessageProducer does not support setting the channel by name.");
+	}
+
 	/**
 	 * Return the the output channel.
 	 * @return the channel.

spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.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.
@@ -118,7 +118,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 
 	private static final Set<MessageProducer> REFERENCED_REPLY_PRODUCERS = new HashSet<>();
 
-	protected final Map<Object, String> integrationComponents = new LinkedHashMap<>();
+	protected final Map<Object, String> integrationComponents = new LinkedHashMap<>(); //NOSONAR - final
 
 	private MessageChannel currentMessageChannel;
 
@@ -3062,36 +3062,19 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 					channelName = ((MessageChannelReference) outputChannel).getName();
 				}
 
-				Object currentComponent = this.currentComponent;
-
-				if (AopUtils.isAopProxy(currentComponent)) {
-					currentComponent = extractProxyTarget(currentComponent);
-				}
-
-				if (currentComponent instanceof MessageProducer) {
-					MessageProducer messageProducer =
-							(MessageProducer) currentComponent;
+				if (this.currentComponent instanceof MessageProducer) {
+					MessageProducer messageProducer = (MessageProducer) this.currentComponent;
 					checkReuse(messageProducer);
 					if (channelName != null) {
-						if (messageProducer instanceof AbstractMessageProducingHandler) {
-							((AbstractMessageProducingHandler) messageProducer).setOutputChannelName(channelName);
-						}
-						else {
-							throw new BeanCreationException("The 'currentComponent' (" + currentComponent
-									+ ") must extend 'AbstractMessageProducingHandler' "
-									+ "for message channel resolution by name.\n"
-									+ "Your handler should extend 'AbstractMessageProducingHandler', "
-									+ "its subclass 'AbstractReplyProducingMessageHandler', or you should "
-									+ "reference a 'MessageChannel' bean instead of its name.");
-						}
+						messageProducer.setOutputChannelName(channelName);
 					}
 					else {
 						messageProducer.setOutputChannel(outputChannel);
 					}
 				}
-				else if (currentComponent instanceof SourcePollingChannelAdapterSpec) {
+				else if (this.currentComponent instanceof SourcePollingChannelAdapterSpec) {
 					SourcePollingChannelAdapterFactoryBean pollingChannelAdapterFactoryBean =
-							((SourcePollingChannelAdapterSpec) currentComponent).get().getT1();
+							((SourcePollingChannelAdapterSpec) this.currentComponent).get().getT1();
 					if (channelName != null) {
 						pollingChannelAdapterFactoryBean.setOutputChannelName(channelName);
 					}
@@ -3100,7 +3083,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 					}
 				}
 				else {
-					throw new BeanCreationException("The 'currentComponent' (" + currentComponent +
+					throw new BeanCreationException("The 'currentComponent' (" + this.currentComponent +
 							") is a one-way 'MessageHandler' and it isn't appropriate to configure 'outputChannel'. " +
 							"This is the end of the integration flow.");
 				}

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

@@ -73,6 +73,7 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements
 	 * @param outputChannelName the channel name.
 	 * @since 4.3
 	 */
+	@Override
 	public void setOutputChannelName(String outputChannelName) {
 		Assert.hasText(outputChannelName, "'outputChannelName' must not be null or empty");
 		this.outputChannelName = outputChannelName;

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

@@ -91,6 +91,7 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan
 		this.outputChannel = outputChannel;
 	}
 
+	@Override
 	public void setOutputChannelName(String outputChannelName) {
 		Assert.hasText(outputChannelName, "'outputChannelName' must not be empty");
 		this.outputChannelName = outputChannelName; //NOSONAR (inconsistent sync)