INT-4513: Clean IntManageConf from removed beans

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

The `IntegrationManagementConfigurer` keeps metrics beans in its local
stores.
When we add beans at runtime, they are not removed from those caches
after their removal

* Implement `DestructionAwareBeanPostProcessor` in the `IntegrationManagementConfigurer`
and clean up caches according provided bean type and its name
* Improve `ManualFlowTests` and `FtpTests` to be sure that
`IntegrationManagementConfigurer` caches are cleared after destroying
 `IntegrationFlowRegistration`

**Cherry-pick to 5.0.x**
This commit is contained in:
Artem Bilan
2018-07-27 12:36:39 -04:00
committed by Gary Russell
parent bd0fd50313
commit adbbca13fb
3 changed files with 57 additions and 5 deletions

View File

@@ -27,7 +27,7 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.beans.factory.config.DestructionAwareBeanPostProcessor;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.integration.core.MessageSource;
@@ -64,8 +64,9 @@ import org.springframework.util.StringUtils;
* @since 4.2
*
*/
public class IntegrationManagementConfigurer implements SmartInitializingSingleton, ApplicationContextAware,
BeanNameAware, BeanPostProcessor {
public class IntegrationManagementConfigurer
implements SmartInitializingSingleton, ApplicationContextAware, BeanNameAware,
DestructionAwareBeanPostProcessor {
private static final Log logger = LogFactory.getLog(IntegrationManagementConfigurer.class);
@@ -246,7 +247,8 @@ public class IntegrationManagementConfigurer implements SmartInitializingSinglet
this.metricsFactory = new DefaultMetricsFactory();
}
this.sourceConfigurers.putAll(this.applicationContext.getBeansOfType(MessageSourceMetricsConfigurer.class));
Map<String, IntegrationManagement> managed = this.applicationContext.getBeansOfType(IntegrationManagement.class);
Map<String, IntegrationManagement> managed = this.applicationContext
.getBeansOfType(IntegrationManagement.class);
for (Entry<String, IntegrationManagement> entry : managed.entrySet()) {
IntegrationManagement bean = entry.getValue();
if (!bean.getOverrides().loggingConfigured) {
@@ -259,7 +261,8 @@ public class IntegrationManagementConfigurer implements SmartInitializingSinglet
}
private void injectCaptor() {
Map<String, IntegrationManagement> managed = this.applicationContext.getBeansOfType(IntegrationManagement.class);
Map<String, IntegrationManagement> managed = this.applicationContext
.getBeansOfType(IntegrationManagement.class);
for (Entry<String, IntegrationManagement> entry : managed.entrySet()) {
IntegrationManagement bean = entry.getValue();
if (!bean.getOverrides().loggingConfigured) {
@@ -280,6 +283,30 @@ public class IntegrationManagementConfigurer implements SmartInitializingSinglet
return bean;
}
@Override
public boolean requiresDestruction(Object bean) {
return bean instanceof MessageChannelMetrics ||
bean instanceof MessageHandlerMetrics ||
bean instanceof MessageSourceMetrics;
}
@Override
public void postProcessBeforeDestruction(Object bean, String beanName) throws BeansException {
if (bean instanceof MessageChannelMetrics) {
this.channelsByName.remove(beanName);
}
else if (bean instanceof MessageHandlerMetrics) {
if (this.handlersByName.remove(beanName) == null) {
this.handlersByName.remove(beanName + ".handler");
}
}
else if (bean instanceof MessageSourceMetrics) {
if (this.sourcesByName.remove(beanName) == null) {
this.sourcesByName.remove(beanName + ".source");
}
}
}
private Object doConfigureMetrics(Object bean, String name) {
if (bean instanceof MessageChannelMetrics) {
configureChannelMetrics(name, (MessageChannelMetrics) bean);

View File

@@ -60,7 +60,9 @@ import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.config.EnableIntegrationManagement;
import org.springframework.integration.config.EnableMessageHistory;
import org.springframework.integration.config.IntegrationManagementConfigurer;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.dsl.IntegrationFlow;
@@ -114,6 +116,9 @@ public class ManualFlowTests {
@Autowired
private SmartLifecycleRoleController roleController;
@Autowired
private IntegrationManagementConfigurer integrationManagementConfigurer;
@Test
@SuppressWarnings("unchecked")
public void testWithAnonymousMessageProducerStart() {
@@ -128,9 +133,11 @@ public class ManualFlowTests {
};
QueueChannel channel = new QueueChannel();
channel.setBeanName("channel");
IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(producer);
BridgeHandler bridgeHandler = new BridgeHandler();
bridgeHandler.setBeanName("bridge");
IntegrationFlow flow =
flowBuilder.handle(bridgeHandler)
@@ -145,9 +152,15 @@ public class ManualFlowTests {
assertTrue(replyProducers.contains(bridgeHandler));
assertNotNull(this.integrationManagementConfigurer.getChannelMetrics("channel"));
assertNotNull(this.integrationManagementConfigurer.getHandlerMetrics("bridge"));
flowRegistration.destroy();
assertFalse(replyProducers.contains(bridgeHandler));
assertNull(this.integrationManagementConfigurer.getChannelMetrics("channel"));
assertNull(this.integrationManagementConfigurer.getHandlerMetrics("bridge"));
}
@Test
@@ -504,6 +517,7 @@ public class ManualFlowTests {
@Configuration
@EnableIntegration
@EnableMessageHistory
@EnableIntegrationManagement
public static class RootConfiguration {
@Bean

View File

@@ -46,6 +46,8 @@ import org.springframework.context.annotation.Configuration;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.config.EnableIntegrationManagement;
import org.springframework.integration.config.IntegrationManagementConfigurer;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
@@ -85,6 +87,9 @@ public class FtpTests extends FtpTestSupport {
@Autowired
private ApplicationContext context;
@Autowired
private IntegrationManagementConfigurer integrationManagementConfigurer;
@Test
public void testFtpInboundFlow() throws IOException {
QueueChannel out = new QueueChannel();
@@ -132,7 +137,12 @@ public class FtpTests extends FtpTestSupport {
MessageSource<?> source = context.getBean(FtpInboundFileSynchronizingMessageSource.class);
assertThat(TestUtils.getPropertyValue(source, "maxFetchSize"), equalTo(10));
assertNotNull(this.integrationManagementConfigurer.getSourceMetrics("ftpInboundAdapter.source"));
registration.destroy();
assertNull(this.integrationManagementConfigurer.getSourceMetrics("ftpInboundAdapter.source"));
}
@Test
@@ -221,6 +231,7 @@ public class FtpTests extends FtpTestSupport {
@Configuration
@EnableIntegration
@EnableIntegrationManagement
public static class ContextConfiguration {
}