Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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

Browse Source 
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
This commit is contained in:
Artem Bilan
2018-02-21 16:07:36 -05:00
committed by Gary Russell
parent f6084939cc
commit 7d1680534f
3 changed files with 74 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<B extends IntegrationFlowDefinit
* {@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 abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
* 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 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
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 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
}
}
if (componentsToRegister != null) {
componentsToRegister.clear();
}
register(routerSpec, null);
if (routerSpec.isDefaultToParentFlow()) {
routerSpec.defaultOutputChannel(new FixedSubscriberChannel(bridgeHandler));
registerSubflowBridge = true;

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

@@ -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 @@ package org.springframework.integration.dsl;
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 final class RouterSpec<K, R extends AbstractMappingMessageRouter>
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 @@ public final class RouterSpec<K, R extends AbstractMappingMessageRouter>
}
@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();
}
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<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());
}
this.router.setChannelMapping(channelKey, entry.getValue().getComponentName());
}
}

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

@@ -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<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 {