INTEXT-113: Make .route() as non-last component

JIRA: https://jira.spring.io/browse/INTEXT-113

Previously `router` could be configured only as last component in the `IntegrationFlow` definition.

* Remove `defaultOutputChannel` options from `AbstractRouterSpec`
* Populate `outputChannel` to the `AbstractMessageRouter.defaultOutputChannel`, when there is more EIP-methods after `.route()`
* Rename EIP-method `.recipientListRoute()` to the `.routeToRecipients()`
* Make compatibility with SI 4.1
* Tested against SF 4.1

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

@@ -18,7 +18,6 @@ package org.springframework.integration.dsl;
 
 import org.springframework.integration.dsl.core.MessageHandlerSpec;
 import org.springframework.integration.router.AbstractMessageRouter;
-import org.springframework.messaging.MessageChannel;
 
 /**
  * @author Artem Bilan
@@ -30,16 +29,6 @@ public class AbstractRouterSpec<S extends AbstractRouterSpec<S, R>, R extends Ab
 		this.target = router;
 	}
 
-	public S defaultOutputChannel(MessageChannel defaultOutputChannel) {
-		this.target.setDefaultOutputChannel(defaultOutputChannel);
-		return _this();
-	}
-
-	public S defaultOutputChannel(String defaultOutputChannel) {
-		this.target.setDefaultOutputChannelName(defaultOutputChannel);
-		return _this();
-	}
-
 	public S ignoreSendFailures(boolean ignoreSendFailures) {
 		this.target.setIgnoreSendFailures(ignoreSendFailures);
 		return _this();

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

@@ -29,6 +29,7 @@ import org.springframework.integration.expression.ValueExpression;
 import org.springframework.integration.store.MessageGroupStore;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.scheduling.TaskScheduler;
+import org.springframework.util.StringUtils;
 
 /**
  * @author Artem Bilan
@@ -144,7 +145,9 @@ public abstract class CorrelationHandlerSpec<S extends CorrelationHandlerSpec<S,
 		if (this.discardChannel != null) {
 			handler.setDiscardChannel(this.discardChannel);
 		}
-		handler.setDiscardChannelName(this.discardChannelName);
+		if (StringUtils.hasText(this.discardChannelName)) {
+			handler.setDiscardChannelName(this.discardChannelName);
+		}
 		if (this.messageStore != null) {
 			handler.setMessageStore(this.messageStore);
 		}

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

@@ -38,15 +38,15 @@ class DslRecipientListRouter extends RecipientListRouter {
 
 	private final Map<String, MessageSelector> selectorRecipientMap = new HashMap<String, MessageSelector>();
 
-	void addRecipient(String channelName, String expression) {
+	void add(String channelName, String expression) {
 		this.expressionRecipientMap.put(channelName, expression);
 	}
 
-	void addRecipient(String channelName, MessageSelector selector) {
+	void add(String channelName, MessageSelector selector) {
 		this.selectorRecipientMap.put(channelName, selector);
 	}
 
-	Map<String, Object> getRecipients() {
+	Map<String, Object> get() {
 		Map<String, Object> recipients =
 				new HashMap<String, Object>(this.expressionRecipientMap.size() + this.selectorRecipientMap.size());
 		recipients.putAll(this.expressionRecipientMap);

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

@@ -535,17 +535,17 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 		return this.route(router, endpointConfigurer);
 	}
 
-	public B recipientListRoute(ComponentConfigurer<RecipientListRouterSpec> routerConfigurer) {
-		return this.recipientListRoute(routerConfigurer, null);
+	public B routeToRecipients(ComponentConfigurer<RecipientListRouterSpec> routerConfigurer) {
+		return this.routeToRecipients(routerConfigurer, null);
 	}
 
-	public B recipientListRoute(ComponentConfigurer<RecipientListRouterSpec> routerConfigurer,
+	public B routeToRecipients(ComponentConfigurer<RecipientListRouterSpec> routerConfigurer,
 			EndpointConfigurer<GenericEndpointSpec<RecipientListRouter>> endpointConfigurer) {
 		Assert.notNull(routerConfigurer);
 		RecipientListRouterSpec spec = new RecipientListRouterSpec();
 		routerConfigurer.configure(spec);
 		DslRecipientListRouter recipientListRouter = (DslRecipientListRouter) spec.get();
-		Assert.notEmpty(recipientListRouter.getRecipients(), "recipient list must not be empty");
+		Assert.notEmpty(recipientListRouter.get(), "recipient list must not be empty");
 		return this.route(recipientListRouter, endpointConfigurer);
 	}
 
@@ -644,6 +644,15 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 						messageProducer.setOutputChannel(outputChannel);
 					}
 				}
+				else if (this.currentComponent instanceof AbstractMessageRouter) {
+					AbstractMessageRouter router = (AbstractMessageRouter) this.currentComponent;
+					if (channelName != null) {
+						router.setDefaultOutputChannelName(channelName);
+					}
+					else {
+						router.setDefaultOutputChannel(outputChannel);
+					}
+				}
 				else {
 					throw new BeanCreationException("The 'currentComponent' (" + this.currentComponent +
 							") is a one-way 'MessageHandler' and it isn't appropriate to configure 'outputChannel'. " +

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

@@ -31,13 +31,13 @@ public class RecipientListRouterSpec extends AbstractRouterSpec<RecipientListRou
 
 	public RecipientListRouterSpec recipient(String channelName, String expression) {
 		Assert.hasText(channelName);
-		((DslRecipientListRouter) this.target).addRecipient(channelName, expression);
+		((DslRecipientListRouter) this.target).add(channelName, expression);
 		return _this();
 	}
 
 	public RecipientListRouterSpec recipient(String channelName, MessageSelector selector) {
 		Assert.hasText(channelName);
-		((DslRecipientListRouter) this.target).addRecipient(channelName, selector);
+		((DslRecipientListRouter) this.target).add(channelName, selector);
 		return _this();
 	}
 

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/Files.java

@@ -16,23 +16,24 @@
 
 package org.springframework.integration.dsl.file;
 
+import java.io.File;
 import java.util.Comparator;
 
 /**
  * @author Artem Bilan
  */
-public abstract class File {
+public abstract class Files {
 
-	public static FileInboundChannelAdapterSpec inboundAdapter(java.io.File directory) {
+	public static FileInboundChannelAdapterSpec inboundAdapter(File directory) {
 		return inboundAdapter(directory, null);
 	}
 
-	public static FileInboundChannelAdapterSpec inboundAdapter(java.io.File directory,
-			Comparator<java.io.File> receptionOrderComparator) {
+	public static FileInboundChannelAdapterSpec inboundAdapter(File directory,
+			Comparator<File> receptionOrderComparator) {
 		return new FileInboundChannelAdapterSpec(receptionOrderComparator).directory(directory);
 	}
 
-	public static FileWritingMessageHandlerSpec outboundAdapter(java.io.File destinationDirectory) {
+	public static FileWritingMessageHandlerSpec outboundAdapter(File destinationDirectory) {
 		return new FileWritingMessageHandlerSpec(destinationDirectory).expectReply(false);
 	}
 
@@ -40,7 +41,7 @@ public abstract class File {
 		return new FileWritingMessageHandlerSpec(directoryExpression).expectReply(false);
 	}
 
-	public static FileWritingMessageHandlerSpec outboundGateway(java.io.File destinationDirectory) {
+	public static FileWritingMessageHandlerSpec outboundGateway(File destinationDirectory) {
 		return new FileWritingMessageHandlerSpec(destinationDirectory).expectReply(true);
 	}
 
@@ -48,7 +49,7 @@ public abstract class File {
 		return new FileWritingMessageHandlerSpec(directoryExpression).expectReply(true);
 	}
 
-	public static TailAdapterSpec tailAdapter(java.io.File file) {
+	public static TailAdapterSpec tailAdapter(File file) {
 		return new TailAdapterSpec().file(file);
 	}
 

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

@@ -29,6 +29,7 @@ import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
+import java.io.File;
 import java.io.FileInputStream;
 import java.io.FileOutputStream;
 import java.io.IOException;
@@ -58,15 +59,6 @@ import org.junit.BeforeClass;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
-import com.jcraft.jsch.ChannelSftp;
-import com.mongodb.MongoClient;
-import de.flapdoodle.embed.mongo.MongodExecutable;
-import de.flapdoodle.embed.mongo.MongodStarter;
-import de.flapdoodle.embed.mongo.config.MongodConfigBuilder;
-import de.flapdoodle.embed.mongo.config.Net;
-import de.flapdoodle.embed.mongo.distribution.Version;
-import de.flapdoodle.embed.process.runtime.Network;
-
 import org.springframework.amqp.core.AmqpTemplate;
 import org.springframework.amqp.core.AnonymousQueue;
 import org.springframework.amqp.core.Queue;
@@ -110,7 +102,7 @@ import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.dsl.amqp.Amqp;
 import org.springframework.integration.dsl.channel.DirectChannelSpec;
 import org.springframework.integration.dsl.channel.MessageChannels;
-import org.springframework.integration.dsl.file.File;
+import org.springframework.integration.dsl.file.Files;
 import org.springframework.integration.dsl.ftp.Ftp;
 import org.springframework.integration.dsl.jms.Jms;
 import org.springframework.integration.dsl.sftp.Sftp;
@@ -164,6 +156,15 @@ import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 import org.springframework.util.StreamUtils;
 
+import com.jcraft.jsch.ChannelSftp;
+import com.mongodb.MongoClient;
+import de.flapdoodle.embed.mongo.MongodExecutable;
+import de.flapdoodle.embed.mongo.MongodStarter;
+import de.flapdoodle.embed.mongo.config.MongodConfigBuilder;
+import de.flapdoodle.embed.mongo.config.Net;
+import de.flapdoodle.embed.mongo.distribution.Version;
+import de.flapdoodle.embed.process.runtime.Network;
+
 /**
  * @author Artem Bilan
  * @author Tim Ysewyn
@@ -173,7 +174,7 @@ import org.springframework.util.StreamUtils;
 @DirtiesContext
 public class IntegrationFlowTests {
 
-	private static final java.io.File tmpDir = new java.io.File(System.getProperty("java.io.tmpdir"));
+	private static final File tmpDir = new File(System.getProperty("java.io.tmpdir"));
 
 	private static int mongoPort;
 
@@ -452,22 +453,20 @@ public class IntegrationFlowTests {
 		assertTrue(this.delayedAdvice.getInvoked());
 	}
 
+	@Autowired
+	@Qualifier("routerAsNonLastFlow.input")
+	private MessageChannel routerAsNonLastFlowChannel;
+
+	@Autowired
+	@Qualifier("routerAsNonLastDefaultOutputChannel")
+	private PollableChannel routerAsNonLastDefaultOutputChannel;
+
 	@Test
-	public void testWrongLastComponent() {
-		ConfigurableApplicationContext context = null;
-		try {
-			context = new AnnotationConfigApplicationContext(InvalidLastComponentFlowContext.class);
-			fail("BeanCreationException expected");
-		}
-		catch (Exception e) {
-			assertThat(e, instanceOf(BeanCreationException.class));
-			assertThat(e.getMessage(), containsString("is a one-way 'MessageHandler'"));
-		}
-		finally {
-			if (context != null) {
-				context.close();
-			}
-		}
+	public void testRouterAsNonLastComponent() {
+		this.routerAsNonLastFlowChannel.send(new GenericMessage<String>("Hello World"));
+		Message<?> receive = this.routerAsNonLastDefaultOutputChannel.receive(1000);
+		assertNotNull(receive);
+		assertEquals("Hello World", receive.getPayload());
 	}
 
 	@Test
@@ -514,7 +513,7 @@ public class IntegrationFlowTests {
 		dfa.setPropertyValue("fileNameGenerator", fileNameGenerator);
 		this.fileFlow1Input.send(message);
 
-		assertTrue(new java.io.File(tmpDir, "foo").exists());
+		assertTrue(new File(tmpDir, "foo").exists());
 	}
 
 	@Test
@@ -940,7 +939,7 @@ public class IntegrationFlowTests {
 
 	@Test
 	public void testMessageProducerFlow() throws Exception {
-		FileOutputStream file = new FileOutputStream(new java.io.File(tmpDir, "TailTest"));
+		FileOutputStream file = new FileOutputStream(new File(tmpDir, "TailTest"));
 		for (int i = 0; i < 50; i++) {
 			file.write((i + "\n").getBytes());
 		}
@@ -1078,7 +1077,7 @@ public class IntegrationFlowTests {
 			if (even) {
 				evens.add(i);
 			}
-			FileOutputStream file = new FileOutputStream(new java.io.File(tmpDir, i + extension));
+			FileOutputStream file = new FileOutputStream(new File(tmpDir, i + extension));
 			file.write(("" + i).getBytes());
 			file.flush();
 			file.close();
@@ -1109,8 +1108,8 @@ public class IntegrationFlowTests {
 		this.fileWritingInput.send(new GenericMessage<>(payload));
 		Message<?> receive = this.fileWritingResultChannel.receive(1000);
 		assertNotNull(receive);
-		assertThat(receive.getPayload(), instanceOf(java.io.File.class));
-		java.io.File resultFile = (java.io.File) receive.getPayload();
+		assertThat(receive.getPayload(), instanceOf(File.class));
+		File resultFile = (File) receive.getPayload();
 		assertThat(resultFile.getAbsolutePath(),
 				endsWith(TestUtils.applySystemFileSeparator("fileWritingFlow/foo.sitest")));
 		String fileContent = StreamUtils.copyToString(new FileInputStream(resultFile), Charset.defaultCharset());
@@ -1148,14 +1147,14 @@ public class IntegrationFlowTests {
 		Message<?> message = this.ftpInboundResultChannel.receive(1000);
 		assertNotNull(message);
 		Object payload = message.getPayload();
-		assertThat(payload, instanceOf(java.io.File.class));
-		java.io.File file = (java.io.File) payload;
+		assertThat(payload, instanceOf(File.class));
+		File file = (File) payload;
 		assertThat(file.getName(), isOneOf("FTPSOURCE1.TXT.a", "FTPSOURCE2.TXT.a"));
 		assertThat(file.getAbsolutePath(), containsString("ftpTest"));
 
 		message = this.ftpInboundResultChannel.receive(1000);
 		assertNotNull(message);
-		file = (java.io.File) message.getPayload();
+		file = (File) message.getPayload();
 		assertThat(file.getName(), isOneOf("FTPSOURCE1.TXT.a", "FTPSOURCE2.TXT.a"));
 		assertThat(file.getAbsolutePath(), containsString("ftpTest"));
 
@@ -1171,14 +1170,14 @@ public class IntegrationFlowTests {
 		Message<?> message = this.sftpInboundResultChannel.receive(1000);
 		assertNotNull(message);
 		Object payload = message.getPayload();
-		assertThat(payload, instanceOf(java.io.File.class));
-		java.io.File file = (java.io.File) payload;
+		assertThat(payload, instanceOf(File.class));
+		File file = (File) payload;
 		assertThat(file.getName(), isOneOf("SFTPSOURCE1.TXT.a", "SFTPSOURCE2.TXT.a"));
 		assertThat(file.getAbsolutePath(), containsString("sftpTest"));
 
 		message = this.sftpInboundResultChannel.receive(1000);
 		assertNotNull(message);
-		file = (java.io.File) message.getPayload();
+		file = (File) message.getPayload();
 		assertThat(file.getName(), isOneOf("SFTPSOURCE1.TXT.a", "SFTPSOURCE2.TXT.a"));
 		assertThat(file.getAbsolutePath(), containsString("sftpTest"));
 
@@ -1236,15 +1235,15 @@ public class IntegrationFlowTests {
 		this.ftpMgetInputChannel.send(new GenericMessage<Object>(dir + "*"));
 		Message<?> result = this.remoteFileOutputChannel.receive(1000);
 		assertNotNull(result);
-		List<java.io.File> localFiles = (List<java.io.File>) result.getPayload();
+		List<File> localFiles = (List<File>) result.getPayload();
 		// should have filtered ftpSource2.txt
 		assertEquals(2, localFiles.size());
 
-		for (java.io.File file : localFiles) {
-			assertThat(file.getPath().replaceAll(Matcher.quoteReplacement(java.io.File.separator), "/"),
+		for (File file : localFiles) {
+			assertThat(file.getPath().replaceAll(Matcher.quoteReplacement(File.separator), "/"),
 					Matchers.containsString(dir));
 		}
-		assertThat(localFiles.get(1).getPath().replaceAll(Matcher.quoteReplacement(java.io.File.separator), "/"),
+		assertThat(localFiles.get(1).getPath().replaceAll(Matcher.quoteReplacement(File.separator), "/"),
 				Matchers.containsString(dir + "subFtpSource"));
 	}
 
@@ -1260,15 +1259,15 @@ public class IntegrationFlowTests {
 		this.sftpMgetInputChannel.send(new GenericMessage<Object>(dir + "*"));
 		Message<?> result = this.remoteFileOutputChannel.receive(1000);
 		assertNotNull(result);
-		List<java.io.File> localFiles = (List<java.io.File>) result.getPayload();
+		List<File> localFiles = (List<File>) result.getPayload();
 		// should have filtered sftpSource2.txt
 		assertEquals(2, localFiles.size());
 
-		for (java.io.File file : localFiles) {
-			assertThat(file.getPath().replaceAll(Matcher.quoteReplacement(java.io.File.separator), "/"),
+		for (File file : localFiles) {
+			assertThat(file.getPath().replaceAll(Matcher.quoteReplacement(File.separator), "/"),
 					Matchers.containsString(dir));
 		}
-		assertThat(localFiles.get(1).getPath().replaceAll(Matcher.quoteReplacement(java.io.File.separator), "/"),
+		assertThat(localFiles.get(1).getPath().replaceAll(Matcher.quoteReplacement(File.separator), "/"),
 				Matchers.containsString(dir + "subSftpSource"));
 	}
 
@@ -1471,6 +1470,12 @@ public class IntegrationFlowTests {
 					.get();
 		}
 
+		@Bean
+		public IntegrationFlow routerAsNonLastFlow() {
+			return f -> f.<String, String>route(p -> p, r -> r.resolutionRequired(false))
+					.channel(MessageChannels.queue("routerAsNonLastDefaultOutputChannel"));
+		}
+
 	}
 
 	@Configuration
@@ -1633,13 +1638,11 @@ public class IntegrationFlowTests {
 		public IntegrationFlow recipientListFlow() {
 			return IntegrationFlows.from("recipientListInput")
 					.<String, String>transform(p -> p.replaceFirst("Payload", ""))
-					.recipientListRoute(r ->
-									r.defaultOutputChannel(this.defaultOutputChannel())
-											.recipient("foo-channel", "'foo' == payload")
-											.recipient("bar-channel", m ->
-													m.getHeaders().containsKey("recipient")
-															&& (boolean) m.getHeaders().get("recipient"))
-					)
+					.routeToRecipients(r -> r.recipient("foo-channel", "'foo' == payload")
+							.recipient("bar-channel", m ->
+									m.getHeaders().containsKey("recipient")
+											&& (boolean) m.getHeaders().get("recipient")))
+					.channel("defaultOutputChannel")
 					.get();
 		}
 	}
@@ -1667,7 +1670,7 @@ public class IntegrationFlowTests {
 		@Bean
 		public IntegrationFlow fileFlow1() {
 			return IntegrationFlows.from("fileFlow1Input")
-					.handle(File.outboundAdapter(tmpDir).fileNameGenerator(message -> null),
+					.handle(Files.outboundAdapter(tmpDir).fileNameGenerator(message -> null),
 							c -> c.id("fileWriting"))
 					.get();
 		}
@@ -1846,7 +1849,7 @@ public class IntegrationFlowTests {
 
 		@Bean
 		public IntegrationFlow tailFlow() {
-			return IntegrationFlows.from(File.tailAdapter(new java.io.File(tmpDir, "TailTest"))
+			return IntegrationFlows.from(Files.tailAdapter(new File(tmpDir, "TailTest"))
 					.delay(500)
 					.id("tailer"))
 					.transform("hello "::concat)
@@ -1925,7 +1928,7 @@ public class IntegrationFlowTests {
 		@Bean
 		public IntegrationFlow fileReadingFlow() {
 			return IntegrationFlows
-					.from(File.inboundAdapter(tmpDir).patternFilter("*.sitest"),
+					.from(Files.inboundAdapter(tmpDir).patternFilter("*.sitest"),
 							e -> e.poller(Pollers.fixedDelay(100)))
 					.transform(Transformers.fileToString())
 					.aggregate(a -> a.correlationExpression("1")
@@ -1938,8 +1941,8 @@ public class IntegrationFlowTests {
 		public IntegrationFlow fileWritingFlow() {
 			return IntegrationFlows.from("fileWritingInput")
 					.enrichHeaders(h -> h.header(FileHeaders.FILENAME, "foo.sitest")
-							.header("directory", new java.io.File(tmpDir, "fileWritingFlow")))
-					.handle(File.outboundGateway("headers[directory]"))
+							.header("directory", new File(tmpDir, "fileWritingFlow")))
+					.handle(Files.outboundGateway("headers[directory]"))
 					.channel(MessageChannels.queue("fileWritingResultChannel"))
 					.get();
 		}
@@ -1988,18 +1991,6 @@ public class IntegrationFlowTests {
 	}
 
 
-	private static class InvalidLastComponentFlowContext {
-
-		@Bean
-		public IntegrationFlow wrongLastComponent() {
-			return IntegrationFlows.from(MessageChannels.direct())
-					.route(Object::toString)
-					.channel(MessageChannels.direct())
-					.get();
-		}
-
-	}
-
 	private static class InvalidLastMessageChannelFlowContext {
 
 		@Bean