INT-4411: DSL: Fix sub-flows for dynamic routers

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

The `RouterSpec.RouterMappingProvider` relies on the
`ContextRefreshedEvent` which happens only during application start up.
When we register `IntegrationFlow` at runtime, this event doesn't
happen and therefore sub-flow mappings don't populated.

* Fix `RouterSpec.RouterMappingProvider` to parse sub-flow mappings in
the `onInit()`
* Reorder components registration for the router in the
`IntegrationFlowDefinition` to let lately sub-flows to start earlier,
then lifecycles in the main flow

Author: artembilan

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2018 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.
@@ -535,7 +535,7 @@ public <S, T> B transform(GenericTransformer<S, T> genericTransformer) {
 	 * {@link org.springframework.integration.handler.MessageProcessor} from provided {@link MessageProcessorSpec}.
 	 * <pre class="code">
 	 * {@code
-	 *  .transform(Scripts.script("classpath:myScript.py").valiable("foo", bar()))
+	 *  .transform(Scripts.script("classpath:myScript.py").variable("foo", bar()))
 	 * }
 	 * </pre>
 	 * @param messageProcessorSpec the {@link MessageProcessorSpec} to use.
@@ -552,7 +552,7 @@ public B transform(MessageProcessorSpec<?> messageProcessorSpec) {
 	 * In addition accept options for the integration endpoint using {@link GenericEndpointSpec}.
 	 * <pre class="code">
 	 * {@code
-	 *  .transform(Scripts.script("classpath:myScript.py").valiable("foo", bar()),
+	 *  .transform(Scripts.script("classpath:myScript.py").variable("foo", bar()),
 	 *           e -> e.autoStartup(false))
 	 * }
 	 * </pre>
@@ -1973,7 +1973,16 @@ private <R extends AbstractMessageRouter, S extends AbstractRouterSpec<S, R>> B
 
 		BridgeHandler bridgeHandler = new BridgeHandler();
 		boolean registerSubflowBridge = false;
-		Map<Object, String> componentsToRegister = routerSpec.getComponentsToRegister();
+
+		Map<Object, String> componentsToRegister = null;
+		Map<Object, String> routerComponents = routerSpec.getComponentsToRegister();
+		if (routerComponents != null) {
+			componentsToRegister = new LinkedHashMap<>(routerComponents);
+			routerComponents.clear();
+		}
+
+		register(routerSpec, null);
+
 		if (!CollectionUtils.isEmpty(componentsToRegister)) {
 			for (Map.Entry<Object, String> entry : componentsToRegister.entrySet()) {
 				Object component = entry.getKey();
@@ -1991,12 +2000,6 @@ private <R extends AbstractMessageRouter, S extends AbstractRouterSpec<S, R>> B
 			}
 		}
 
-		if (componentsToRegister != null) {
-			componentsToRegister.clear();
-		}
-
-		register(routerSpec, null);
-
 		if (routerSpec.isDefaultToParentFlow()) {
 			routerSpec.defaultOutputChannel(new FixedSubscriberChannel(bridgeHandler));
 			registerSubflowBridge = true;

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2018 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.
@@ -18,10 +18,7 @@
 
 import java.util.HashMap;
 import java.util.Map;
-import java.util.concurrent.atomic.AtomicBoolean;
 
-import org.springframework.context.ApplicationListener;
-import org.springframework.context.event.ContextRefreshedEvent;
 import org.springframework.core.convert.ConversionService;
 import org.springframework.core.convert.support.DefaultConversionService;
 import org.springframework.integration.channel.DirectChannel;
@@ -179,14 +176,11 @@ public Map<Object, String> getComponentsToRegister() {
 		return super.getComponentsToRegister();
 	}
 
-	private static class RouterMappingProvider extends IntegrationObjectSupport
-			implements ApplicationListener<ContextRefreshedEvent> {
-
-		private final AtomicBoolean initialized = new AtomicBoolean();
+	private static class RouterMappingProvider extends IntegrationObjectSupport {
 
 		private final MappingMessageRouterManagement router;
 
-		private final Map<Object, NamedComponent> mapping = new HashMap<Object, NamedComponent>();
+		private final Map<Object, NamedComponent> mapping = new HashMap<>();
 
 		RouterMappingProvider(MappingMessageRouterManagement router) {
 			this.router = router;
@@ -197,31 +191,30 @@ void addMapping(Object key, NamedComponent channel) {
 		}
 
 		@Override
-		public void onApplicationEvent(ContextRefreshedEvent event) {
-			if (event.getApplicationContext() == getApplicationContext() && !this.initialized.getAndSet(true)) {
-				ConversionService conversionService = getConversionService();
-				if (conversionService == null) {
-					conversionService = DefaultConversionService.getSharedInstance();
+		protected void onInit() throws Exception {
+			super.onInit();
+			ConversionService conversionService = getConversionService();
+			if (conversionService == null) {
+				conversionService = DefaultConversionService.getSharedInstance();
+			}
+			for (Map.Entry<Object, NamedComponent> entry : this.mapping.entrySet()) {
+				Object key = entry.getKey();
+				String channelKey;
+				if (key instanceof String) {
+					channelKey = (String) key;
+				}
+				else if (key instanceof Class) {
+					channelKey = ((Class<?>) key).getName();
 				}
-				for (Map.Entry<Object, NamedComponent> entry : this.mapping.entrySet()) {
-					Object key = entry.getKey();
-					String channelKey;
-					if (key instanceof String) {
-						channelKey = (String) key;
-					}
-					else if (key instanceof Class) {
-						channelKey = ((Class<?>) key).getName();
-					}
-					else if (conversionService.canConvert(key.getClass(), String.class)) {
-						channelKey = conversionService.convert(key, String.class);
-					}
-					else {
-						throw new MessagingException("Unsupported channel mapping type for router ["
-								+ key.getClass() + "]");
-					}
-
-					this.router.setChannelMapping(channelKey, entry.getValue().getComponentName());
+				else if (conversionService.canConvert(key.getClass(), String.class)) {
+					channelKey = conversionService.convert(key, String.class);
 				}
+				else {
+					throw new MessagingException("Unsupported channel mapping type for router ["
+							+ key.getClass() + "]");
+				}
+
+				this.router.setChannelMapping(channelKey, entry.getValue().getComponentName());
 			}
 		}
 

spring-integration-core/src/test/java/org/springframework/integration/dsl/manualflow/ManualFlowTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2018 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.
@@ -21,11 +21,13 @@
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
+import java.util.Arrays;
 import java.util.Date;
 import java.util.Objects;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -70,6 +72,8 @@
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringRunner;
 
+import reactor.core.publisher.Flux;
+
 /**
  * @author Artem Bilan
  * @author Gary Russell
@@ -356,6 +360,37 @@ public void testRoleControl() {
 		assertTrue(this.roleController.getEndpointsRunningStatus(testRole).isEmpty());
 	}
 
+	@Test
+	public void testDynaSubFlowCreation() {
+		Flux<Message<?>> messageFlux =
+				Flux.just("1,2,3,4")
+						.map(v -> v.split(","))
+						.flatMapIterable(Arrays::asList)
+						.map(Integer::parseInt)
+						.map(GenericMessage::new);
+
+		QueueChannel resultChannel = new QueueChannel();
+
+		IntegrationFlow integrationFlow = IntegrationFlows
+				.from(messageFlux)
+				.<Integer, Boolean>route(p -> p % 2 == 0, m -> m
+						.subFlowMapping(true, sf -> sf.<Integer, String>transform(em -> "even:" + em))
+						.subFlowMapping(false, sf -> sf.<Integer, String>transform(em -> "odd:" + em))
+						.defaultOutputToParentFlow()
+				)
+				.channel(resultChannel)
+				.get();
+
+		this.integrationFlowContext.registration(integrationFlow).register();
+
+		for (int i = 0; i < 4; i++) {
+			Message<?> receive = resultChannel.receive(10_000);
+			assertNotNull(receive);
+		}
+
+		assertNull(resultChannel.receive(0));
+	}
+
 	@Configuration
 	@EnableIntegration
 	public static class RootConfiguration {