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

INT-3455 Orderly Shutdown Improvements

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3455

Do not stop schedulers and executors - allows mid-flow
QueueChannels to be drained.

Stop all inbound MessageProducers (that are not OrderlyShutdownCapable).

INT-3455 Polishing; PR Comments

Change deprecate method usage to the new version
This commit is contained in:
Gary Russell
2014-08-05 22:17:50 +03:00
committed by Artem Bilan
parent 5f214ea559
commit 6b8bdc7e5e
8 changed files with 162 additions and 230 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MBeanExporterIntegrationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2009-2012 the original author or authors.
* Copyright 2009-2014 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. You may obtain a copy of the License at
@@ -28,17 +28,22 @@ import javax.management.ObjectName;
import org.junit.After;
import org.junit.Test;
import org.springframework.aop.framework.Advised;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.Lifecycle;
import org.springframework.context.support.GenericXmlApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.context.OrderlyShutdownCapable;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.jmx.export.annotation.ManagedResource;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.util.Assert;
public class MBeanExporterIntegrationTests {
@@ -100,7 +105,7 @@ public class MBeanExporterIntegrationTests {
assertNotNull(messageChannelsMonitor);
MBeanServer server = context.getBean(MBeanServer.class);
Set<ObjectName> names = server.queryNames(ObjectName.getInstance("org.springframework.integration:type=ManagedEndpoint,*"), null);
assertEquals(0, names.size());
assertEquals(2, names.size());
names = server.queryNames(ObjectName.getInstance("org.springframework.integration:name=explicit,*"), null);
assertEquals(1, names.size());
MBeanOperationInfo[] operations = server.getMBeanInfo(names.iterator().next()).getOperations();
@@ -114,13 +119,39 @@ public class MBeanExporterIntegrationTests {
// Lifecycle method name
assertEquals("start", startName);
assertTrue((Boolean) server.invoke(names.iterator().next(), "isRunning", null, null));
messageChannelsMonitor.stopActiveComponents(false, 3000);
messageChannelsMonitor.stopActiveComponents(3000);
assertFalse((Boolean) server.invoke(names.iterator().next(), "isRunning", null, null));
ActiveChannel activeChannel = context.getBean("activeChannel", ActiveChannel.class);
assertTrue(activeChannel.isStopCalled());
OtherActiveComponent otherActiveComponent = context.getBean(OtherActiveComponent.class);
assertTrue(otherActiveComponent.isBeforeCalled());
assertTrue(otherActiveComponent.isAfterCalled());
assertTrue(otherActiveComponent.isRunning());
assertFalse(context.getBean(AMessageProducer.class).isRunning());
// check pollers are still running
QueueChannel input = (QueueChannel) extractTarget(context.getBean("input"));
QueueChannel input2 = (QueueChannel) extractTarget(context.getBean("input2"));
input.purge(null);
input2.purge(null);
input.send(new GenericMessage<String>("foo"));
assertNotNull(input2.receive(10000));
}
private Object extractTarget(Object bean) {
if (!(bean instanceof Advised)) {
return bean;
}
Advised advised = (Advised) bean;
if (advised.getTargetSource() == null) {
return null;
}
try {
return extractTarget(advised.getTargetSource().getTarget());
}
catch (Exception e) {
return null;
}
}
@Test
@@ -214,15 +245,18 @@ public class MBeanExporterIntegrationTests {
// This has the potential to blow up if called before setDate().
// Depends on bean instantiation order and IntegrationMBeanExporter
// can influence that by aggressively instantiating other MBeanExporters
@Override
public Date getObject() throws Exception {
Assert.state(date != null, "A date must be provided");
return date;
}
@Override
public Class<?> getObjectType() {
return Date.class;
}
@Override
public boolean isSingleton() {
return true;
}
@@ -237,6 +271,7 @@ public class MBeanExporterIntegrationTests {
this.date = date;
}
@Override
public void afterPropertiesSet() throws Exception {
Assert.state(date != null, "A date must be provided");
}
@@ -246,21 +281,24 @@ public class MBeanExporterIntegrationTests {
public static class MetricFactoryBean implements FactoryBean<Metric> {
private MessageChannel channel;
private Metric date = new Metric();
private final Metric date = new Metric();
public void setChannel(MessageChannel channel) {
this.channel = channel;
}
@Override
public Metric getObject() throws Exception {
Assert.state(channel != null, "A channel must be provided");
return date;
}
@Override
public Class<?> getObjectType() {
return Metric.class;
}
@Override
public boolean isSingleton() {
return true;
}
@@ -280,6 +318,7 @@ public class MBeanExporterIntegrationTests {
this.channel = channel;
}
@Override
public void afterPropertiesSet() throws Exception {
Assert.state(channel != null, "A channel must be provided");
}
@@ -294,12 +333,14 @@ public class MBeanExporterIntegrationTests {
public static class SimpleService implements Service {
private int counter;
@Override
public String execute() throws Exception {
Thread.sleep(10L); // make the duration non-zero
counter++;
return "count="+counter;
}
@Override
public int getCounter() {
return counter;
}
@@ -313,31 +354,38 @@ public class MBeanExporterIntegrationTests {
private boolean stopCalled;
@Override
public boolean send(Message<?> message) {
return false;
}
@Override
public boolean send(Message<?> message, long timeout) {
return false;
}
@Override
public void start() {
}
@Override
public void stop() {
this.stopCalled = true;
}
@Override
public boolean isRunning() {
return false;
}
@Override
public boolean isStopCalled() {
return this.stopCalled;
}
}
public static class OtherActiveComponent implements OrderlyShutdownCapable {
public static class OtherActiveComponent extends MessageProducerSupport
implements OrderlyShutdownCapable {
private boolean beforeCalled;
@@ -351,14 +399,20 @@ public class MBeanExporterIntegrationTests {
return afterCalled;
}
@Override
public int beforeShutdown() {
this.beforeCalled = true;
return 0;
}
@Override
public int afterShutdown() {
this.afterCalled = true;
return 0;
}
}
public static class AMessageProducer extends MessageProducerSupport {
}
}

16
spring-integration-jmx/src/test/java/org/springframework/integration/monitor/lifecycle-source.xml
View File

@@ -15,6 +15,14 @@
<int:queue />
</int:channel>
<int:bridge input-channel="input" output-channel="input2">
<int:poller fixed-delay="200" />
</int:bridge>
<int:channel id="input2">
<int:queue />
</int:channel>
<int:inbound-channel-adapter id="explicit" ref="service" method="execute" channel="input">
<int:poller fixed-rate="200" />
</int:inbound-channel-adapter>
@@ -29,7 +37,13 @@
<bean id="nonSpringExecutor" class="java.util.concurrent.Executors" factory-method="newSingleThreadExecutor" />
<bean id="otherActiveComponent" class="org.springframework.integration.monitor.MBeanExporterIntegrationTests$OtherActiveComponent" />
<bean id="otherActiveComponent" class="org.springframework.integration.monitor.MBeanExporterIntegrationTests$OtherActiveComponent">
<property name="outputChannel" ref="input" />
</bean>
<bean id="messageProducer" class="org.springframework.integration.monitor.MBeanExporterIntegrationTests$AMessageProducer">
<property name="outputChannel" ref="input" />
</bean>
<bean id="ignoreWrappedExecutor" class="org.springframework.core.task.support.ExecutorServiceAdapter">
<constructor-arg ref="someExecutor" />

8
spring-integration-jmx/src/test/java/org/springframework/integration/monitor/self-destruction-context.xml
View File

@@ -18,15 +18,13 @@
<int:control-bus input-channel="toControlBus" />
<int-jmx:mbean-export id="integrationMbeanExporter" shutdown-executor="shutdownExec"
<int-jmx:mbean-export id="integrationMbeanExporter"
default-domain="self-destruct" />
<context:mbean-server />
<int:poller default="true" fixed-delay="500" />
<task:executor id="shutdownExec" />
<bean id="activeChannel" class="org.springframework.integration.monitor.MBeanExporterIntegrationTests$ActiveChannelImpl" />
<task:scheduler id="someScheduler" />
@@ -35,6 +33,8 @@
<bean id="nonSpringExecutor" class="java.util.concurrent.Executors" factory-method="newSingleThreadExecutor" />
<bean id="otherActiveComponent" class="org.springframework.integration.monitor.MBeanExporterIntegrationTests$OtherActiveComponent" />
<bean id="otherActiveComponent" class="org.springframework.integration.monitor.MBeanExporterIntegrationTests$OtherActiveComponent">
<property name="outputChannel" ref="nullChannel" />
</bean>
</beans>