Add support for FixedSubscriberChannel

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java

@@ -19,12 +19,14 @@ package org.springframework.integration.dsl;
 import org.springframework.beans.factory.BeanCreationException;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.FixedSubscriberChannel;
 import org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean;
 import org.springframework.integration.core.GenericSelector;
 import org.springframework.integration.core.MessageSelector;
 import org.springframework.integration.dsl.channel.MessageChannelSpec;
 import org.springframework.integration.dsl.core.ConsumerEndpointSpec;
-import org.springframework.integration.dsl.core.MessageChannelReference;
+import org.springframework.integration.dsl.support.FixedSubscriberChannelPrototype;
+import org.springframework.integration.dsl.support.MessageChannelReference;
 import org.springframework.integration.dsl.support.BeanNameMethodInvokingMessageHandler;
 import org.springframework.integration.dsl.support.EndpointConfigurer;
 import org.springframework.integration.dsl.support.EnricherConfigurer;
@@ -71,6 +73,14 @@ public final class IntegrationFlowBuilder {
 		return this;
 	}
 
+	public IntegrationFlowBuilder fixedSubscriberChannel() {
+		return this.fixedSubscriberChannel(null);
+	}
+
+	public IntegrationFlowBuilder fixedSubscriberChannel(String messageChannelName) {
+		return this.channel(new FixedSubscriberChannelPrototype(messageChannelName));
+	}
+
 	public IntegrationFlowBuilder channel(String messageChannelName) {
 		return this.channel(new MessageChannelReference(messageChannelName));
 	}
@@ -78,14 +88,7 @@ public final class IntegrationFlowBuilder {
 	public IntegrationFlowBuilder channel(MessageChannel messageChannel) {
 		Assert.notNull(messageChannel);
 		if (this.currentMessageChannel != null) {
-			GenericEndpointSpec<BridgeHandler> endpointSpec = new GenericEndpointSpec<BridgeHandler>(new BridgeHandler());
-			if (this.currentMessageChannel instanceof MessageChannelReference) {
-				endpointSpec.get().getT1().setInputChannelName(((MessageChannelReference) this.currentMessageChannel).getName());
-			}
-			else {
-				endpointSpec.get().getT1().setInputChannel(this.currentMessageChannel);
-			}
-			this.addComponent(endpointSpec).currentComponent(endpointSpec.get().getT2());
+			this.register(new GenericEndpointSpec<BridgeHandler>(new BridgeHandler()), null);
 		}
 		this.currentMessageChannel = messageChannel;
 		return this.registerOutputChannelIfCan(this.currentMessageChannel);
@@ -188,6 +191,14 @@ public final class IntegrationFlowBuilder {
 			endpointSpec.get().getT1().setInputChannelName(((MessageChannelReference) inputChannel).getName());
 		}
 		else {
+			if (inputChannel instanceof FixedSubscriberChannelPrototype) {
+				String beanName = ((FixedSubscriberChannelPrototype) inputChannel).getName();
+				inputChannel = new FixedSubscriberChannel(endpointSpec.get().getT2());
+				if (beanName != null) {
+					((FixedSubscriberChannel) inputChannel).setBeanName(beanName);
+				}
+				this.registerOutputChannelIfCan(inputChannel);
+			}
 			endpointSpec.get().getT1().setInputChannel(inputChannel);
 		}
 
@@ -195,40 +206,47 @@ public final class IntegrationFlowBuilder {
 	}
 
 	private IntegrationFlowBuilder registerOutputChannelIfCan(MessageChannel outputChannel) {
-		this.flow.addComponent(outputChannel);
-		String channelName = null;
-		if (outputChannel instanceof MessageChannelReference) {
-			channelName = ((MessageChannelReference) outputChannel).getName();
-		}
-		if (this.currentComponent != null) {
-			if (this.currentComponent instanceof AbstractReplyProducingMessageHandler) {
-				AbstractReplyProducingMessageHandler messageProducer = (AbstractReplyProducingMessageHandler) this.currentComponent;
-				if (channelName != null) {
-					messageProducer.setOutputChannelName(channelName);
+		if (!(outputChannel instanceof FixedSubscriberChannelPrototype)) {
+			this.flow.addComponent(outputChannel);
+			if (this.currentComponent != null) {
+				String channelName = null;
+				if (outputChannel instanceof MessageChannelReference) {
+					channelName = ((MessageChannelReference) outputChannel).getName();
+				}
+				if (this.currentComponent instanceof AbstractReplyProducingMessageHandler) {
+					AbstractReplyProducingMessageHandler messageProducer = (AbstractReplyProducingMessageHandler) this.currentComponent;
+					if (channelName != null) {
+						messageProducer.setOutputChannelName(channelName);
+					}
+					else {
+						messageProducer.setOutputChannel(outputChannel);
+					}
+				}
+				else if (this.currentComponent instanceof SourcePollingChannelAdapterFactoryBean) {
+					SourcePollingChannelAdapterFactoryBean pollingChannelAdapterFactoryBean = (SourcePollingChannelAdapterFactoryBean) this.currentComponent;
+					if (channelName != null) {
+						pollingChannelAdapterFactoryBean.setOutputChannelName(channelName);
+					}
+					else {
+						pollingChannelAdapterFactoryBean.setOutputChannel(outputChannel);
+					}
 				}
 				else {
-					messageProducer.setOutputChannel(outputChannel);
+					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.");
 				}
+				this.currentComponent = null;
 			}
-			else if (this.currentComponent instanceof SourcePollingChannelAdapterFactoryBean) {
-				SourcePollingChannelAdapterFactoryBean pollingChannelAdapterFactoryBean = (SourcePollingChannelAdapterFactoryBean) this.currentComponent;
-				if (channelName != null) {
-					pollingChannelAdapterFactoryBean.setOutputChannelName(channelName);
-				}
-				else {
-					pollingChannelAdapterFactoryBean.setOutputChannel(outputChannel);
-				}
-			}
-			else {
-				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.");
-			}
-			this.currentComponent = null;
 		}
 		return this;
 	}
 
 	public IntegrationFlow get() {
+		if (this.currentMessageChannel instanceof FixedSubscriberChannelPrototype) {
+			throw new BeanCreationException("The 'currentMessageChannel' (" + this.currentMessageChannel + ") is a prototype" +
+					" for FixedSubscriberChannel which can't be created without MessageHandler constructor argument. " +
+					"That means that '.fixedSubscriberChannel()' can't be the last EIP-method in the IntegrationFlow definition.");
+		}
 		return this.flow;
 	}
 

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java

@@ -19,7 +19,8 @@ package org.springframework.integration.dsl;
 import org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean;
 import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.dsl.channel.MessageChannelSpec;
-import org.springframework.integration.dsl.core.MessageChannelReference;
+import org.springframework.integration.dsl.support.FixedSubscriberChannelPrototype;
+import org.springframework.integration.dsl.support.MessageChannelReference;
 import org.springframework.integration.dsl.support.EndpointConfigurer;
 import org.springframework.messaging.MessageChannel;
 
@@ -28,10 +29,25 @@ import org.springframework.messaging.MessageChannel;
  */
 public final class IntegrationFlows {
 
+	/**
+	 * @param messageChannelName the name of existing {@link org.springframework.messaging.MessageChannel} bean.
+	 *                           The new {@link org.springframework.integration.channel.DirectChannel} bean will be
+	 *                           created on context startup, if there is on bean with this name.
+	 * @return new {@link IntegrationFlowBuilder}
+	 */
 	public static IntegrationFlowBuilder from(String messageChannelName) {
 		return from(new MessageChannelReference(messageChannelName));
 	}
 
+	/**
+	 * @param messageChannelName the name for {@link org.springframework.integration.channel.FixedSubscriberChannel}
+	 *                           to be created on context startup, not reference.
+	 * @return new {@link IntegrationFlowBuilder}
+	 */
+	public static IntegrationFlowBuilder fromFixedMessageChannel(String messageChannelName) {
+		return from(new FixedSubscriberChannelPrototype(messageChannelName));
+	}
+
 	public static IntegrationFlowBuilder from(MessageChannel messageChannel) {
 		return new IntegrationFlowBuilder().channel(messageChannel);
 	}

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

@@ -34,11 +34,13 @@ import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
 import org.springframework.beans.factory.support.BeanDefinitionRegistry;
 import org.springframework.beans.factory.support.GenericBeanDefinition;
 import org.springframework.integration.channel.AbstractMessageChannel;
+import org.springframework.integration.channel.FixedSubscriberChannel;
 import org.springframework.integration.config.ConsumerEndpointFactoryBean;
 import org.springframework.integration.config.IntegrationConfigUtils;
 import org.springframework.integration.config.IntegrationConfigurationInitializer;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.config.InstanceBeanDefinition;
+import org.springframework.integration.dsl.support.MessageChannelReference;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.util.Assert;
 
@@ -121,6 +123,14 @@ public class DslIntegrationConfigurationInitializer implements IntegrationConfig
 								IntegrationConfigUtils.autoCreateDirectChannel(channelName, registry);
 							}
 						}
+						else if (instance instanceof FixedSubscriberChannel) {
+							FixedSubscriberChannel fixedSubscriberChannel = (FixedSubscriberChannel) instance;
+							String channelBeanName = fixedSubscriberChannel.getComponentName();
+							if ("Unnamed fixed subscriber channel".equals(channelBeanName)) {
+								channelBeanName = flowNamePrefix + "channel" + BeanFactoryUtils.GENERATED_BEAN_NAME_SEPARATOR + channelNameIndex++;
+							}
+							registry.registerBeanDefinition(channelBeanName, component);
+						}
 						else {
 							String beanName = generateInstanceBeanDefinitionName(registry, instance);
 							registry.registerBeanDefinition(beanName, component);

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/FixedSubscriberChannelPrototype.java

@@ -0,0 +1,44 @@
+package org.springframework.integration.dsl.support;
+
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+
+/**
+ * @author Artem Bilan
+ * @since 4.0
+ */
+public class FixedSubscriberChannelPrototype implements MessageChannel {
+
+	private final String name;
+
+	public FixedSubscriberChannelPrototype() {
+		this(null);
+	}
+
+	public FixedSubscriberChannelPrototype(String name) {
+		this.name = name;
+	}
+
+	public String getName() {
+		return name;
+	}
+
+
+	@Override
+	public boolean send(Message<?> message) {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public boolean send(Message<?> message, long timeout) {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public String toString() {
+		return "FixedSubscriberChannelPrototype{" +
+				"name='" + name + '\'' +
+				'}';
+	}
+
+}

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/MessageChannelReference.java

@@ -1,4 +1,4 @@
-package org.springframework.integration.dsl.core;
+package org.springframework.integration.dsl.support;
 
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;

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

@@ -52,6 +52,7 @@ import org.springframework.integration.MessageDispatchingException;
 import org.springframework.integration.annotation.MessageEndpoint;
 import org.springframework.integration.annotation.ServiceActivator;
 import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.FixedSubscriberChannel;
 import org.springframework.integration.channel.PublishSubscribeChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.config.EnableIntegration;
@@ -154,11 +155,12 @@ public class IntegrationFlowTests {
 
 	@Autowired
 	@Qualifier("enricherInput")
-	private DirectChannel enricherInput;
+	private FixedSubscriberChannel enricherInput;
 
 
 	@Test
 	public void testPollingFlow() {
+		assertThat(this.beanFactory.getBean("integerChannel"), Matchers.instanceOf(FixedSubscriberChannel.class));
 		for (int i = 0; i < 10; i++) {
 			Message<?> message = this.outputChannel.receive(5000);
 			assertNotNull(message);
@@ -215,6 +217,9 @@ public class IntegrationFlowTests {
 		assertNotNull(reply);
 		assertEquals("test", reply.getPayload());
 
+		assertTrue(this.beanFactory.containsBean("bridgeFlow2:channel#0"));
+		assertThat(this.beanFactory.getBean("bridgeFlow2:channel#0"), Matchers.instanceOf(FixedSubscriberChannel.class));
+
 		try {
 			this.bridgeFlow2Input.send(message);
 			fail("Expected MessageDispatchingException");
@@ -244,6 +249,18 @@ public class IntegrationFlowTests {
 		}
 	}
 
+	@Test
+	public void testWrongLastMessageChannel() {
+		try {
+			new AnnotationConfigApplicationContext(InvalidLastMessageChannelFlowContext.class);
+			fail("BeanCreationException expected");
+		}
+		catch (Exception e) {
+			assertThat(e, Matchers.instanceOf(BeanCreationException.class));
+			assertThat(e.getMessage(), Matchers.containsString("'.fixedSubscriberChannel()' can't be the last EIP-method in the IntegrationFlow definition"));
+		}
+	}
+
 
 	@Test
 	public void testFileHandler() {
@@ -318,7 +335,7 @@ public class IntegrationFlowTests {
 		@Bean
 		public IntegrationFlow flow1() {
 			return IntegrationFlows.from(this.integerMessageSource(), c -> c.poller(Pollers.fixedRate(100)))
-					.channel("integerChannel")
+					.fixedSubscriberChannel("integerChannel")
 					.transform("payload.toString()")
 					.channel(MessageChannels.queue("flow1QueueChannel"))
 					.get();
@@ -367,6 +384,7 @@ public class IntegrationFlowTests {
 			return IntegrationFlows.from(this.inputChannel)
 					.filter(p -> p instanceof String, c -> c.id("filter"))
 					.channel("foo")
+					.fixedSubscriberChannel()
 					.<String, Integer>transform(Integer::parseInt)
 					.transform(new PayloadSerializingTransformer(),
 							c -> c.autoStartup(false).id("payloadSerializingTransformer"))
@@ -434,6 +452,7 @@ public class IntegrationFlowTests {
 		public IntegrationFlow bridgeFlow2() {
 			return IntegrationFlows.from("bridgeFlow2Input")
 					.bridge(c -> c.autoStartup(false).id("bridge"))
+					.fixedSubscriberChannel()
 					.delay("delayer", "200", c -> c.advice(this.delayedAdvice))
 					.channel(MessageChannels.queue("bridgeFlow2Output"))
 					.get();
@@ -486,7 +505,7 @@ public class IntegrationFlowTests {
 
 		@Bean
 		public IntegrationFlow enricherFlow() {
-			return IntegrationFlows.from("enricherInput")
+			return IntegrationFlows.fromFixedMessageChannel("enricherInput")
 					.enrich(e -> e.requestChannel("enrichChannel")
 									.requestPayloadExpression("payload")
 									.shouldClonePayload(false)
@@ -527,6 +546,17 @@ public class IntegrationFlowTests {
 
 	}
 
+	private static class InvalidLastMessageChannelFlowContext {
+
+		@Bean
+		public IntegrationFlow wrongLastComponent() {
+			return IntegrationFlows.from(MessageChannels.direct())
+					.fixedSubscriberChannel()
+					.get();
+		}
+
+	}
+
 	@EnableIntegration
 	public static class InvalidConfigurationWithSpec {
 

spring-integration-java-dsl/src/test/resources/log4j.properties

@@ -4,5 +4,4 @@ log4j.appender.stdout=org.apache.log4j.ConsoleAppender
 log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
 log4j.appender.stdout.layout.ConversionPattern=%d{HH:mm:ss.SSS} %-5p [%t][%c] %m%n
 
-log4j.category.org.springframework.integration=WARN
 log4j.category.org.springframework.integration.dsl=INFO