Remove dependant beans recursively

https://build.spring.io/browse/INT-MASTER-1058/

When we destroy manually registered `IntegrationFlow`, we need to
count with sub-flows and iterate all the dependant beans recursively
for full flow removal

**Cherry-pick to 5.0.x**
This commit is contained in:
Artem Bilan
2018-05-31 16:28:46 -04:00
committed by Gary Russell
parent ba962fcd15
commit 88945994fa
2 changed files with 25 additions and 16 deletions

View File

@@ -16,7 +16,6 @@
package org.springframework.integration.dsl.context;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
@@ -77,6 +76,8 @@ public final class IntegrationFlowContext implements BeanFactoryAware {
private ConfigurableListableBeanFactory beanFactory;
private BeanDefinitionRegistry beanDefinitionRegistry;
private IntegrationFlowContext() {
}
@@ -87,6 +88,7 @@ public final class IntegrationFlowContext implements BeanFactoryAware {
"'ConfigurableListableBeanFactory'. " +
"Consider using 'GenericApplicationContext' implementation.");
this.beanFactory = (ConfigurableListableBeanFactory) beanFactory;
this.beanDefinitionRegistry = (BeanDefinitionRegistry) this.beanFactory;
}
/**
@@ -184,17 +186,9 @@ public final class IntegrationFlowContext implements BeanFactoryAware {
IntegrationFlowRegistration flowRegistration = this.registry.remove(flowId);
flowRegistration.stop();
BeanDefinitionRegistry beanDefinitionRegistry = (BeanDefinitionRegistry) this.beanFactory;
removeDependantBeans(flowId);
Arrays.stream(this.beanFactory.getDependentBeans(flowId))
.forEach(beanName -> {
beanDefinitionRegistry.removeBeanDefinition(beanName);
// TODO until https://jira.spring.io/browse/SPR-16837
Arrays.asList(beanDefinitionRegistry.getAliases(beanName))
.forEach(beanDefinitionRegistry::removeAlias);
});
beanDefinitionRegistry.removeBeanDefinition(flowId);
this.beanDefinitionRegistry.removeBeanDefinition(flowId);
}
else {
throw new IllegalStateException("An IntegrationFlow with the id "
@@ -202,6 +196,19 @@ public final class IntegrationFlowContext implements BeanFactoryAware {
}
}
private void removeDependantBeans(String parentName) {
String[] dependentBeans = this.beanFactory.getDependentBeans(parentName);
for (String beanName : dependentBeans) {
removeDependantBeans(beanName);
this.beanDefinitionRegistry.removeBeanDefinition(beanName);
// TODO until https://jira.spring.io/browse/SPR-16837
String[] aliases = this.beanDefinitionRegistry.getAliases(beanName);
for (String alias : aliases) {
this.beanDefinitionRegistry.removeAlias(alias);
}
}
}
/**
* Obtain a {@link MessagingTemplate} with its default destination set to the input channel
* of the {@link IntegrationFlow} for provided {@code flowId}.
@@ -249,7 +256,7 @@ public final class IntegrationFlowContext implements BeanFactoryAware {
*/
public final class IntegrationFlowRegistrationBuilder {
private Map<Object, String> additionalBeans = new HashMap<>();
private final Map<Object, String> additionalBeans = new HashMap<>();
private final IntegrationFlowRegistration integrationFlowRegistration;

View File

@@ -28,11 +28,11 @@ import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
@@ -392,7 +392,7 @@ public class ManualFlowTests {
assertTrue(this.roleController.getEndpointsRunningStatus(testRole).isEmpty());
}
// @Test
@Test
public void testDynamicSubFlowCreation() {
Flux<Message<?>> messageFlux =
Flux.just("1,2,3,4")
@@ -414,7 +414,9 @@ public class ManualFlowTests {
.get();
IntegrationFlowRegistration flowRegistration =
this.integrationFlowContext.registration(integrationFlow).register();
this.integrationFlowContext.registration(integrationFlow)
.id("dynamicSubFlows")
.register();
for (int i = 0; i < 4; i++) {
Message<?> receive = resultChannel.receive(10_000);
@@ -458,7 +460,7 @@ public class ManualFlowTests {
public void testConcurrentRegistration() throws InterruptedException {
ExecutorService executorService = Executors.newCachedThreadPool();
List<IntegrationFlowRegistration> flowRegistrations = new ArrayList<>();
List<IntegrationFlowRegistration> flowRegistrations = new CopyOnWriteArrayList<>();
AtomicBoolean exceptionHappened = new AtomicBoolean();