INT-4539: Fix Java DSL for prototype beans

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

The `IntegrationFlowBeanPostProcessor` doesn't check for prototype beans
and just override them in the application context with the singletons

* Since we can't in the DSL understand if provided object is a prototype
or not, we try to check for its bean definition by possible bean name.
Use `NamedComponent` for possible bean name to check.
* Remove `final` from the `IntegrationComponentSpec.get()` since its
result is not visible for CGI proxies when it is declared as a `@Bean`
and subsequent `getObject()` produces a new internal object
* Verify prototype beans with new test in the `IntegrationFlowTests`

**Cherry-pick to 5.0.x**

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

@@ -63,7 +63,7 @@ public abstract class IntegrationComponentSpec<S extends IntegrationComponentSpe
 	/**
 	 * @return the configured component.
 	 */
-	public final T get() {
+	public T get() {
 		if (this.target == null) {
 			this.target = doGet();
 		}
@@ -71,7 +71,7 @@ public abstract class IntegrationComponentSpec<S extends IntegrationComponentSpe
 	}
 
 	@Override
-	public T getObject() throws Exception {
+	public T getObject() {
 		return get();
 	}
 
@@ -80,11 +80,6 @@ public abstract class IntegrationComponentSpec<S extends IntegrationComponentSpe
 		return get().getClass();
 	}
 
-	@Override
-	public boolean isSingleton() {
-		return true;
-	}
-
 	@Override
 	public void afterPropertiesSet() throws Exception {
 		if (this.target instanceof InitializingBean) {

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

@@ -163,11 +163,7 @@ public class IntegrationFlowBeanPostProcessor
 					id = flowNamePrefix + id;
 				}
 
-				Collection<?> messageHandlers =
-						this.beanFactory.getBeansOfType(messageHandler.getClass(), false, false)
-								.values();
-
-				if (!messageHandlers.contains(messageHandler)) {
+				if (noBeanPresentForComponent(messageHandler)) {
 					String handlerBeanName = generateBeanName(messageHandler, flowNamePrefix);
 
 					registerComponent(messageHandler, handlerBeanName, flowBeanName);
@@ -178,8 +174,7 @@ public class IntegrationFlowBeanPostProcessor
 				targetIntegrationComponents.put(endpoint, id);
 			}
 			else {
-				Collection<?> values = this.beanFactory.getBeansOfType(component.getClass(), false, false).values();
-				if (!values.contains(component)) {
+				if (noBeanPresentForComponent(component)) {
 					if (component instanceof AbstractMessageChannel) {
 						String channelBeanName = ((AbstractMessageChannel) component).getComponentName();
 						if (channelBeanName == null) {
@@ -216,10 +211,7 @@ public class IntegrationFlowBeanPostProcessor
 						if (!CollectionUtils.isEmpty(componentsToRegister)) {
 							componentsToRegister.entrySet()
 									.stream()
-									.filter(o ->
-											!this.beanFactory.getBeansOfType(o.getKey().getClass(), false, false)
-													.values()
-													.contains(o.getKey()))
+									.filter(o -> noBeanPresentForComponent(o.getKey()))
 									.forEach(o ->
 											registerComponent(o.getKey(),
 													generateBeanName(o.getKey(), flowNamePrefix, o.getValue(),
@@ -239,9 +231,7 @@ public class IntegrationFlowBeanPostProcessor
 						targetIntegrationComponents.put(pollingChannelAdapterFactoryBean, id);
 
 						MessageSource<?> messageSource = spec.get().getT2();
-						if (!this.beanFactory.getBeansOfType(messageSource.getClass(), false, false)
-								.values()
-								.contains(messageSource)) {
+						if (noBeanPresentForComponent(messageSource)) {
 							String messageSourceId = id + ".source";
 							if (messageSource instanceof NamedComponent
 									&& ((NamedComponent) messageSource).getComponentName() != null) {
@@ -347,10 +337,7 @@ public class IntegrationFlowBeanPostProcessor
 
 				componentsToRegister.entrySet()
 						.stream()
-						.filter(component ->
-								!this.beanFactory.getBeansOfType(component.getKey().getClass(), false, false)
-										.values()
-										.contains(component.getKey()))
+						.filter(component -> noBeanPresentForComponent(component.getKey()))
 						.forEach(component ->
 								registerComponent(component.getKey(),
 										generateBeanName(component.getKey(), component.getValue())));
@@ -391,6 +378,19 @@ public class IntegrationFlowBeanPostProcessor
 		}
 	}
 
+	private boolean noBeanPresentForComponent(Object instance) {
+		if (instance instanceof NamedComponent) {
+			String beanName = ((NamedComponent) instance).getComponentName();
+			if (beanName != null) {
+				return !this.beanFactory.containsBean(beanName);
+			}
+		}
+
+		Collection<?> beans = this.beanFactory.getBeansOfType(instance.getClass(), false, false).values();
+
+		return !beans.contains(instance);
+	}
+
 	private void registerComponent(Object component, String beanName) {
 		registerComponent(component, beanName, null);
 	}

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

@@ -20,6 +20,7 @@ import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNotSame;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertThat;
@@ -44,11 +45,13 @@ import org.springframework.beans.factory.BeanCreationException;
 import org.springframework.beans.factory.ListableBeanFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.beans.factory.config.ConfigurableBeanFactory;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.ComponentScan;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Scope;
 import org.springframework.integration.MessageDispatchingException;
 import org.springframework.integration.MessageRejectedException;
 import org.springframework.integration.annotation.MessageEndpoint;
@@ -65,6 +68,7 @@ import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.dsl.MessageChannels;
 import org.springframework.integration.dsl.Pollers;
 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.advice.ErrorMessageSendingRecoverer;
@@ -499,6 +503,18 @@ public class IntegrationFlowTests {
 		this.nullChannel.setCountsEnabled(false);
 	}
 
+	@Autowired
+	private EventDrivenConsumer flow1WithPrototypeHandlerConsumer;
+
+	@Autowired
+	private EventDrivenConsumer flow2WithPrototypeHandlerConsumer;
+
+	@Test
+	public void testPrototypeIsNotOverridden() {
+		assertNotSame(this.flow1WithPrototypeHandlerConsumer.getHandler(),
+				this.flow2WithPrototypeHandlerConsumer.getHandler());
+	}
+
 	@MessagingGateway
 	public interface ControlBusGateway {
 
@@ -865,6 +881,31 @@ public class IntegrationFlowTests {
 					.nullChannel();
 		}
 
+		@Bean
+		@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
+		public AbstractReplyProducingMessageHandler myHandler() {
+			return new AbstractReplyProducingMessageHandler() {
+
+				@Override
+				protected Object handleRequestMessage(Message<?> requestMessage) {
+					return requestMessage;
+				}
+
+			};
+		}
+
+		@Bean
+		public IntegrationFlow flow1WithPrototypeHandler(
+				@Qualifier("myHandler") AbstractReplyProducingMessageHandler handler) {
+			return f -> f.handle(handler, e -> e.id("flow1WithPrototypeHandlerConsumer"));
+		}
+
+		@Bean
+		public IntegrationFlow flow2WithPrototypeHandler(
+				@Qualifier("myHandler") AbstractReplyProducingMessageHandler handler) {
+			return f -> f.handle(handler, e -> e.id("flow2WithPrototypeHandlerConsumer"));
+		}
+
 	}
 
 	@Service