From 7d1680534f844519c018ae899af2f67924c9379c Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 21 Feb 2018 16:07:36 -0500 Subject: [PATCH] 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 --- .../dsl/IntegrationFlowDefinition.java | 23 ++++---- .../integration/dsl/RouterSpec.java | 57 ++++++++----------- .../dsl/manualflow/ManualFlowTests.java | 37 +++++++++++- 3 files changed, 74 insertions(+), 43 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java index a0b9c747e1..29b8043412 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java +++ b/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 abstract class IntegrationFlowDefinition * {@code - * .transform(Scripts.script("classpath:myScript.py").valiable("foo", bar())) + * .transform(Scripts.script("classpath:myScript.py").variable("foo", bar())) * } * * @param messageProcessorSpec the {@link MessageProcessorSpec} to use. @@ -552,7 +552,7 @@ public abstract class IntegrationFlowDefinition * {@code - * .transform(Scripts.script("classpath:myScript.py").valiable("foo", bar()), + * .transform(Scripts.script("classpath:myScript.py").variable("foo", bar()), * e -> e.autoStartup(false)) * } * @@ -1973,7 +1973,16 @@ public abstract class IntegrationFlowDefinition componentsToRegister = routerSpec.getComponentsToRegister(); + + Map componentsToRegister = null; + Map routerComponents = routerSpec.getComponentsToRegister(); + if (routerComponents != null) { + componentsToRegister = new LinkedHashMap<>(routerComponents); + routerComponents.clear(); + } + + register(routerSpec, null); + if (!CollectionUtils.isEmpty(componentsToRegister)) { for (Map.Entry entry : componentsToRegister.entrySet()) { Object component = entry.getKey(); @@ -1991,12 +2000,6 @@ public abstract class IntegrationFlowDefinition return super.getComponentsToRegister(); } - private static class RouterMappingProvider extends IntegrationObjectSupport - implements ApplicationListener { - - private final AtomicBoolean initialized = new AtomicBoolean(); + private static class RouterMappingProvider extends IntegrationObjectSupport { private final MappingMessageRouterManagement router; - private final Map mapping = new HashMap(); + private final Map mapping = new HashMap<>(); RouterMappingProvider(MappingMessageRouterManagement router) { this.router = router; @@ -197,31 +191,30 @@ public final class RouterSpec } @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 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() + "]"); } - for (Map.Entry 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()); - } + this.router.setChannelMapping(channelKey, entry.getValue().getComponentName()); } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/manualflow/ManualFlowTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/manualflow/ManualFlowTests.java index 6770f841ca..1ae3bd7ff7 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/manualflow/ManualFlowTests.java +++ b/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.hamcrest.Matchers.instanceOf; 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.annotation.DirtiesContext; 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 class ManualFlowTests { assertTrue(this.roleController.getEndpointsRunningStatus(testRole).isEmpty()); } + @Test + public void testDynaSubFlowCreation() { + Flux> 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) + .route(p -> p % 2 == 0, m -> m + .subFlowMapping(true, sf -> sf.transform(em -> "even:" + em)) + .subFlowMapping(false, sf -> sf.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 {