diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java index 884001244c..d268751432 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java @@ -33,7 +33,7 @@ import org.springframework.util.Assert; /** * Adapter that receives Messages from an AMQP Queue, converts them into * Spring Integration Messages, and sends the results to a Message Channel. - * + * * @author Mark Fisher * @author Gary Russell * @since 2.1 @@ -90,4 +90,24 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements this.messageListenerContainer.stop(); } + + /** + * {@inheritDoc} + *

+ * Shuts down the listener container. + */ + public int beforeShutdown() { + this.stop(); + return 0; + } + + + /** + * {@inheritDoc} + *

No-op + */ + public int afterShutdown() { + return 0; + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/core/OrderlyShutdownCapable.java b/spring-integration-core/src/main/java/org/springframework/integration/core/OrderlyShutdownCapable.java index 53310ed87c..afd270e774 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/core/OrderlyShutdownCapable.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/core/OrderlyShutdownCapable.java @@ -16,15 +16,33 @@ package org.springframework.integration.core; /** - * Marker interface for components that wish to be considered for - * an orderly shutdown using management interfaces. Components that - * implement this interface will be stopped before schedulers, - * executors etc, in order for them to free up any execution - * threads they may be holding. + * Interface for components that wish to be considered for + * an orderly shutdown using management interfaces. beforeShuddown() + * will be called before schedulers, executors etc, are stopped. + * afterShutdown() is called after the shutdown delay. + * * @author Gary Russell * @since 2.2 * */ public interface OrderlyShutdownCapable { + /** + * Called before shutdown begins. Implementations should + * stop accepting new messages. Can optionally return the + * number of active messages in process. + * @return The number of active messages if available. + */ + int beforeShutdown(); + + /** + * Called after normal shutdown of schedulers, executors etc, + * and after the shutdown delay has elapsed, but before any + * forced shutdown of any remaining active scheduler/executor + * threads.Can optionally return the number of active messages + * still in process. + * @return The number of active messages if available. + */ + int afterShutdown(); + } diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/HttpRequestHandlingEndpointSupport.java b/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/HttpRequestHandlingEndpointSupport.java index b2739854e9..f4a61db65e 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/HttpRequestHandlingEndpointSupport.java +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/HttpRequestHandlingEndpointSupport.java @@ -21,6 +21,7 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.List; import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; import javax.servlet.http.HttpServletRequest; import javax.servlet.http.HttpServletResponse; @@ -49,6 +50,7 @@ import org.springframework.http.server.ServletServerHttpResponse; import org.springframework.integration.Message; import org.springframework.integration.MessageHeaders; import org.springframework.integration.MessagingException; +import org.springframework.integration.core.OrderlyShutdownCapable; import org.springframework.integration.gateway.MessagingGatewaySupport; import org.springframework.integration.http.converter.MultipartAwareFormHttpMessageConverter; import org.springframework.integration.http.converter.SerializingHttpMessageConverter; @@ -95,7 +97,8 @@ import org.springframework.web.util.UrlPathHelper; * @author Gary Russell * @since 2.0 */ -abstract class HttpRequestHandlingEndpointSupport extends MessagingGatewaySupport { +abstract class HttpRequestHandlingEndpointSupport extends MessagingGatewaySupport + implements OrderlyShutdownCapable { private static final boolean jaxb2Present = ClassUtils.isPresent("javax.xml.bind.Binder", HttpRequestHandlingEndpointSupport.class.getClassLoader()); @@ -132,6 +135,10 @@ abstract class HttpRequestHandlingEndpointSupport extends MessagingGatewaySuppor private volatile Map headerExpressions; + private volatile boolean shuttingDown; + + private final AtomicInteger activeCount = new AtomicInteger(); + public HttpRequestHandlingEndpointSupport() { this(true); } @@ -264,6 +271,10 @@ abstract class HttpRequestHandlingEndpointSupport extends MessagingGatewaySuppor this.multipartResolver = multipartResolver; } + protected boolean isShuttingDown() { + return this.shuttingDown; + } + @Override public String getComponentType() { return (this.expectReply) ? "http:inbound-gateway" : "http:inbound-channel-adapter"; @@ -298,13 +309,30 @@ abstract class HttpRequestHandlingEndpointSupport extends MessagingGatewaySuppor this.validateSupportedMethods(); } + + @Override + protected void doStart() { + this.shuttingDown = false; + super.doStart(); + } + /** * Handles the HTTP request by generating a Message and sending it to the request channel. If this gateway's * 'expectReply' property is true, it will also generate a response from the reply Message once received. * @return a the response Message */ - @SuppressWarnings({ "rawtypes", "unchecked" }) protected final Message doHandleRequest(HttpServletRequest servletRequest, HttpServletResponse servletResponse) throws IOException { + if (this.isShuttingDown()) { + return createServiceUnavailableResponse(); + } + else { + return actualDoHandleRequest(servletRequest, servletResponse); + } + } + + @SuppressWarnings({ "rawtypes", "unchecked" }) + private Message actualDoHandleRequest(HttpServletRequest servletRequest, HttpServletResponse servletResponse) throws IOException { + this.activeCount.incrementAndGet(); try { ServletServerHttpRequest request = this.prepareRequest(servletRequest); if (!this.supportedMethods.contains(request.getMethod())) { @@ -386,9 +414,22 @@ abstract class HttpRequestHandlingEndpointSupport extends MessagingGatewaySuppor } finally { this.postProcessRequest(servletRequest); + this.activeCount.decrementAndGet(); } } + /** + * @return + */ + private Message createServiceUnavailableResponse() { + if (logger.isDebugEnabled()) { + logger.debug("Endpoint is shutting down; returning status " + HttpStatus.SERVICE_UNAVAILABLE); + } + return MessageBuilder.withPayload("Endpoint is shutting down") + .setHeader(org.springframework.integration.http.HttpHeaders.STATUS_CODE, HttpStatus.SERVICE_UNAVAILABLE) + .build(); + } + /** * Converts the reply message to the appropriate HTTP reply object and * sets up the servlet response. @@ -521,4 +562,13 @@ abstract class HttpRequestHandlingEndpointSupport extends MessagingGatewaySuppor } } } + + public int beforeShutdown() { + this.shuttingDown = true; + return this.activeCount.get(); + } + + public int afterShutdown() { + return this.activeCount.get(); + } } diff --git a/spring-integration-http/src/test/java/org/springframework/integration/http/inbound/HttpRequestHandlingControllerTests.java b/spring-integration-http/src/test/java/org/springframework/integration/http/inbound/HttpRequestHandlingControllerTests.java index d303de6c9f..122b07b3cc 100644 --- a/spring-integration-http/src/test/java/org/springframework/integration/http/inbound/HttpRequestHandlingControllerTests.java +++ b/spring-integration-http/src/test/java/org/springframework/integration/http/inbound/HttpRequestHandlingControllerTests.java @@ -23,9 +23,16 @@ import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; import static org.mockito.Mockito.mock; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + import org.junit.Test; import org.springframework.expression.Expression; import org.springframework.expression.spel.standard.SpelExpressionParser; +import org.springframework.http.HttpStatus; import org.springframework.integration.Message; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; @@ -246,5 +253,69 @@ public class HttpRequestHandlingControllerTests { assertTrue("Wrong message: "+error, ((String)error.getArguments()[1]).startsWith("failed to send Message")); } + @Test + public void shutDown() throws Exception { + DirectChannel requestChannel = new DirectChannel(); + final CountDownLatch latch1 = new CountDownLatch(1); + final CountDownLatch latch2 = new CountDownLatch(1); + AbstractReplyProducingMessageHandler handler = new AbstractReplyProducingMessageHandler() { + @Override + protected Object handleRequestMessage(Message requestMessage) { + try { + latch2.countDown(); + // hold up an active thread so we can verify the count and that it completes ok + latch1.await(10, TimeUnit.SECONDS); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return requestMessage.getPayload().toString().toUpperCase(); + } + }; + requestChannel.subscribe(handler); + final HttpRequestHandlingController controller = new HttpRequestHandlingController(true); + controller.setRequestChannel(requestChannel); + controller.setViewName("foo"); + final MockHttpServletRequest request = new MockHttpServletRequest(); + request.setMethod("POST"); + request.setContent("hello".getBytes()); + request.setContentType("text/plain"); + MockHttpServletResponse response = new MockHttpServletResponse(); + final AtomicInteger active = new AtomicInteger(); + final AtomicBoolean expected503 = new AtomicBoolean(); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + try { + // wait for the active thread + latch2.await(10, TimeUnit.SECONDS); + } + catch (InterruptedException e1) { + Thread.currentThread().interrupt(); + } + // start the shutdown + active.set(controller.beforeShutdown()); + try { + MockHttpServletResponse response = new MockHttpServletResponse(); + controller.handleRequest(request, response); + expected503.set(response.getStatus() == HttpStatus.SERVICE_UNAVAILABLE.value()); + latch1.countDown(); + } + catch (Exception e) { + e.printStackTrace(); + } + } + }); + ModelAndView modelAndView = controller.handleRequest(request, response); + // verify we get a 503 after shutdown starts + assertEquals(1, active.get()); + assertTrue(expected503.get()); + // verify the active request still processed ok + assertEquals("foo", modelAndView.getViewName()); + assertEquals(1, modelAndView.getModel().size()); + Object reply = modelAndView.getModel().get("reply"); + assertNotNull(reply); + assertEquals("HELLO", reply); + } + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java index a32b511e11..c571b06f12 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java @@ -18,8 +18,10 @@ package org.springframework.integration.ip.tcp; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.atomic.AtomicInteger; import org.springframework.integration.Message; +import org.springframework.integration.core.OrderlyShutdownCapable; import org.springframework.integration.gateway.MessagingGatewaySupport; import org.springframework.integration.ip.IpHeaders; import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory; @@ -37,7 +39,7 @@ import org.springframework.util.Assert; /** * Inbound Gateway using a server connection factory - threading is controlled by the * factory. For java.net connections, each socket can process only one message at a time. - * For java.nio connections, messages may be multiplexed but the client will need to + * For java.nio connections, messages may be multiplexed but the client will need to * provide correlation logic. If the client is a {@link TcpOutboundGateway} multiplexing * is not used, but multiple concurrent connections can be used if the connection factory uses * single-use connections. For true asynchronous bi-directional communication, a pair of @@ -47,7 +49,7 @@ import org.springframework.util.Assert; * */ public class TcpInboundGateway extends MessagingGatewaySupport implements - TcpListener, TcpSender, ClientModeCapable { + TcpListener, TcpSender, ClientModeCapable, OrderlyShutdownCapable { private volatile AbstractServerConnectionFactory serverConnectionFactory; @@ -67,7 +69,29 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements private volatile boolean active; + private volatile boolean shuttingDown; + + private final AtomicInteger activeCount = new AtomicInteger(); + public boolean onMessage(Message message) { + if (this.shuttingDown) { + if (logger.isInfoEnabled()) { + logger.info("Inbound message ignored; shutting down; " + message.toString()); + } + } + else { + this.activeCount.incrementAndGet(); + try { + return doOnMessage(message); + } + finally { + this.activeCount.decrementAndGet(); + } + } + return false; + } + + private boolean doOnMessage(Message message) { Message reply = this.sendAndReceiveMessage(message); if (reply == null) { if (logger.isDebugEnabled()) { @@ -92,7 +116,7 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements return false; } - /** + /** * @return true if the associated connection factory is listening. */ public boolean isListening() { @@ -126,6 +150,7 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements public void removeDeadConnection(TcpConnection connection) { connections.remove(connection.getConnectionId()); } + @Override public String getComponentType(){ return "ip:tcp-inbound-gateway"; } @@ -146,6 +171,7 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements super.doStart(); if (!this.active) { this.active = true; + this.shuttingDown = false; if (this.serverConnectionFactory != null) { this.serverConnectionFactory.start(); } @@ -243,4 +269,13 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements } } + public int beforeShutdown() { + this.shuttingDown = true; + return this.activeCount.get(); + } + + public int afterShutdown() { + this.stop(); + return this.activeCount.get(); + } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java index 736d546ef2..e5dcc2c502 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java @@ -16,8 +16,10 @@ package org.springframework.integration.ip.tcp; import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.atomic.AtomicInteger; import org.springframework.integration.Message; +import org.springframework.integration.core.OrderlyShutdownCapable; import org.springframework.integration.endpoint.MessageProducerSupport; import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory; @@ -31,17 +33,17 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.util.Assert; /** - * Tcp inbound channel adapter using a TcpConnection to + * Tcp inbound channel adapter using a TcpConnection to * receive data - if the connection factory is a server * factory, this Listener owns the connections. If it is * a client factory, the sender owns the connection. - * + * * @author Gary Russell * @since 2.0 * */ -public class TcpReceivingChannelAdapter - extends MessageProducerSupport implements TcpListener, ClientModeCapable { +public class TcpReceivingChannelAdapter + extends MessageProducerSupport implements TcpListener, ClientModeCapable, OrderlyShutdownCapable { private AbstractConnectionFactory clientConnectionFactory; @@ -59,8 +61,25 @@ public class TcpReceivingChannelAdapter private volatile boolean active; + private volatile boolean shuttingDown; + + private final AtomicInteger activeCount = new AtomicInteger(); + public boolean onMessage(Message message) { - sendMessage(message); + if (this.shuttingDown) { + if (logger.isInfoEnabled()) { + logger.info("Inbound message ignored; shutting down; " + message.toString()); + } + } + else { + this.activeCount.incrementAndGet(); + try { + sendMessage(message); + } + finally { + this.activeCount.decrementAndGet(); + } + } return false; } @@ -80,6 +99,7 @@ public class TcpReceivingChannelAdapter super.doStart(); if (!this.active) { this.active = true; + this.shuttingDown = false; if (this.serverConnectionFactory != null) { this.serverConnectionFactory.start(); } @@ -117,7 +137,7 @@ public class TcpReceivingChannelAdapter * Sets the client or server connection factory; for this (an inbound adapter), if * the factory is a client connection factory, the sockets are owned by a sending * channel adapter and this adapter is used to receive replies. - * + * * @param connectionFactory the connectionFactory to set */ public void setConnectionFactory(AbstractConnectionFactory connectionFactory) { @@ -126,7 +146,7 @@ public class TcpReceivingChannelAdapter } else { this.serverConnectionFactory = connectionFactory; } - connectionFactory.registerListener(this); + connectionFactory.registerListener(this); } public boolean isListening() { @@ -139,6 +159,7 @@ public class TcpReceivingChannelAdapter return false; } + @Override public String getComponentType(){ return "ip:tcp-inbound-channel-adapter"; } @@ -221,4 +242,13 @@ public class TcpReceivingChannelAdapter } } + public int beforeShutdown() { + this.shuttingDown = true; + return this.activeCount.get(); + } + + public int afterShutdown() { + this.stop(); + return this.activeCount.get(); + } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java index 2d45757d8c..f49048b9e5 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java @@ -40,7 +40,6 @@ import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; import org.springframework.integration.MessagingException; import org.springframework.integration.context.IntegrationObjectSupport; -import org.springframework.integration.core.OrderlyShutdownCapable; import org.springframework.integration.ip.tcp.connection.support.DefaultTcpSocketSupport; import org.springframework.integration.ip.tcp.connection.support.TcpSocketSupport; import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer; @@ -54,7 +53,7 @@ import org.springframework.util.Assert; * */ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport - implements ConnectionFactory, SmartLifecycle, OrderlyShutdownCapable { + implements ConnectionFactory, SmartLifecycle { protected static final int DEFAULT_REPLY_TIMEOUT = 10000; diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java index 5a6266bdba..0b4768b799 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java @@ -20,6 +20,7 @@ import java.net.ServerSocket; import java.net.Socket; import java.net.SocketException; +import org.springframework.integration.core.OrderlyShutdownCapable; import org.springframework.util.Assert; /** @@ -31,7 +32,7 @@ import org.springframework.util.Assert; * @since 2.0 */ public abstract class AbstractServerConnectionFactory - extends AbstractConnectionFactory implements Runnable { + extends AbstractConnectionFactory implements Runnable, OrderlyShutdownCapable { private static final int DEFAULT_BACKLOG = 5; @@ -41,6 +42,8 @@ public abstract class AbstractServerConnectionFactory private volatile int backlog = DEFAULT_BACKLOG; + private volatile boolean shuttingDown; + /** * The port on which the factory will listen. @@ -55,6 +58,7 @@ public abstract class AbstractServerConnectionFactory synchronized (this.lifecycleMonitor) { if (!this.isActive()) { this.setActive(true); + this.shuttingDown = false; this.getTaskExecutor().execute(this); } } @@ -85,6 +89,10 @@ public abstract class AbstractServerConnectionFactory return listening; } + protected boolean isShuttingDown() { + return shuttingDown; + } + /** * Transfers attributes such as (de)serializer, singleUse etc to a new connection. * For single use sockets, enforces a socket timeout (default 10 seconds). @@ -167,4 +175,15 @@ public abstract class AbstractServerConnectionFactory public void setPoolSize(int poolSize) { this.setBacklog(poolSize); } + + public int beforeShutdown() { + this.shuttingDown = true; + return 0; + } + + public int afterShutdown() { + this.stop(); + return 0; + } + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java index 0a5c0dc925..60d4d60530 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java @@ -88,15 +88,24 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto } continue; } - if (logger.isDebugEnabled()) { - logger.debug("Accepted connection from " + socket.getInetAddress().getHostAddress()); + if (this.isShuttingDown()) { + if (logger.isInfoEnabled()) { + logger.info("New connection from " + socket.getInetAddress().getHostAddress() + + " rejected; the server is in the process of shutting down."); + } + socket.close(); + } + else { + if (logger.isDebugEnabled()) { + logger.debug("Accepted connection from " + socket.getInetAddress().getHostAddress()); + } + setSocketAttributes(socket); + TcpConnection connection = new TcpNetConnection(socket, true, this.isLookupHost()); + connection = wrapConnection(connection); + this.initializeConnection(connection, socket); + this.getTaskExecutor().execute(connection); + this.harvestClosedConnections(); } - setSocketAttributes(socket); - TcpConnection connection = new TcpNetConnection(socket, true, this.isLookupHost()); - connection = wrapConnection(connection); - this.initializeConnection(connection, socket); - this.getTaskExecutor().execute(connection); - this.harvestClosedConnections(); } } catch (Exception e) { // don't log an error if we had a good socket once and now it's closed diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java index 8fc4254db0..b6efe7f95d 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java @@ -150,17 +150,26 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto protected void doAccept(final Selector selector, ServerSocketChannel server, long now) throws IOException { logger.debug("New accept"); SocketChannel channel = server.accept(); - channel.configureBlocking(false); - Socket socket = channel.socket(); - setSocketAttributes(socket); - TcpNioConnection connection = createTcpNioConnection(channel); - if (connection == null) { - return; + if (this.isShuttingDown()) { + if (logger.isInfoEnabled()) { + logger.info("New connection from " + channel.socket().getInetAddress().getHostAddress() + + " rejected; the server is in the process of shutting down."); + } + channel.close(); + } + else { + channel.configureBlocking(false); + Socket socket = channel.socket(); + setSocketAttributes(socket); + TcpNioConnection connection = createTcpNioConnection(channel); + if (connection == null) { + return; + } + connection.setTaskExecutor(this.getTaskExecutor()); + connection.setLastRead(now); + this.channelMap.put(channel, connection); + channel.register(selector, SelectionKey.OP_READ, connection); } - connection.setTaskExecutor(this.getTaskExecutor()); - connection.setLastRead(now); - channelMap.put(channel, connection); - channel.register(selector, SelectionKey.OP_READ, connection); } private TcpNioConnection createTcpNioConnection(SocketChannel socketChannel) { diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.java index 008278e32f..c66902a31d 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.java @@ -25,7 +25,7 @@ import org.springframework.util.Assert; /** * A message-driven endpoint that receive JMS messages, converts them into * Spring Integration Messages, and then sends the result to a channel. - * + * * @author Mark Fisher * @author Oleg Zhurakousky * @author Gary Russell @@ -65,13 +65,15 @@ public class JmsMessageDrivenEndpoint extends AbstractEndpoint implements listener.setComponentName(this.getComponentName()); } + @Override protected void doStart() { this.listener.start(); if (!this.listenerContainer.isRunning()) { - this.listenerContainer.start(); + this.listenerContainer.start(); } } + @Override protected void doStop() { this.listenerContainer.stop(); this.listener.stop(); @@ -84,4 +86,15 @@ public class JmsMessageDrivenEndpoint extends AbstractEndpoint implements this.listenerContainer.destroy(); } + + public int beforeShutdown() { + this.stop(); + return 0; + } + + + public int afterShutdown() { + return 0; + } + } diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java index c273666272..3e8e9a3bec 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java @@ -37,7 +37,6 @@ import javax.management.modelmbean.ModelMBean; import org.aopalliance.aop.Advice; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; - import org.springframework.aop.Advisor; import org.springframework.aop.PointcutAdvisor; import org.springframework.aop.TargetSource; @@ -463,7 +462,7 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP */ public void run() { try { - this.stopOrderlyShutdownCapableComponents(); + this.orderlyShutdownCapableComponentsBefore(); this.stopActiveChannels(); this.stopSchedulers(); if (System.currentTimeMillis() > this.shutdownDeadline) { @@ -487,12 +486,14 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP Thread.currentThread().interrupt(); logger.error("Interrupted while waiting for quiesce"); } + this.orderlyShutdownCapableComponentsAfter(); } else { this.shutdownForced = true; this.stopSchedulers(); this.stopExecutors(); this.stopNonSpringExecutors(); + this.orderlyShutdownCapableComponentsAfter(); } } finally { @@ -610,21 +611,32 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP logger.debug("Stopped other executors"); } - @ManagedOperation - public void stopOrderlyShutdownCapableComponents() { - logger.debug("Stopping OrderlyShutdownCapable components"); - Map candidates = this.applicationContext + protected final void orderlyShutdownCapableComponentsBefore() { + logger.debug("Initiating stop OrderlyShutdownCapable components"); + Map components = this.applicationContext .getBeansOfType(OrderlyShutdownCapable.class); - for (Entry candidateEntry : candidates.entrySet()) { - OrderlyShutdownCapable candidate = candidateEntry.getValue(); - if (candidate instanceof Lifecycle) { - if (logger.isInfoEnabled()) { - logger.info("Stopping component " + candidate); - } - ((Lifecycle) candidate).stop(); + for (Entry componentEntry : components.entrySet()) { + OrderlyShutdownCapable component = componentEntry.getValue(); + int n = component.beforeShutdown(); + if (logger.isInfoEnabled()) { + logger.info("Initiated stop for component " + component + "; it reported " + n + " active messages"); } } - logger.debug("Stopped OrderlyShutdownCapable components"); + logger.debug("Initiated stop OrderlyShutdownCapable components"); + } + + protected final void orderlyShutdownCapableComponentsAfter() { + logger.debug("Finalizing stop OrderlyShutdownCapable components"); + Map components = this.applicationContext + .getBeansOfType(OrderlyShutdownCapable.class); + for (Entry componentEntry : components.entrySet()) { + OrderlyShutdownCapable component = componentEntry.getValue(); + int n = component.afterShutdown(); + if (logger.isInfoEnabled()) { + logger.info("Finalized stop for component " + component + "; it reported " + n + " active messages"); + } + } + logger.debug("Initiated stop OrderlyShutdownCapable components"); } @ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Channel Count") diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MBeanExporterIntegrationTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MBeanExporterIntegrationTests.java index 06232281b2..f55051d951 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MBeanExporterIntegrationTests.java +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MBeanExporterIntegrationTests.java @@ -1,11 +1,11 @@ /* * Copyright 2009-2012 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 - * + * * http://www.apache.org/licenses/LICENSE-2.0 - * + * * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the * specific language governing permissions and limitations under the License. @@ -44,9 +44,9 @@ import org.springframework.util.Assert; public class MBeanExporterIntegrationTests { private IntegrationMBeanExporter messageChannelsMonitor; - + private GenericXmlApplicationContext context; - + @After public void close() { if (context!=null) { @@ -55,28 +55,28 @@ public class MBeanExporterIntegrationTests { } @Test - public void testCircularReferenceNoChannel() throws Exception { + public void testCircularReferenceNoChannel() throws Exception { context = new GenericXmlApplicationContext(getClass(), "oref-nonchannel.xml"); messageChannelsMonitor = context.getBean(IntegrationMBeanExporter.class); assertNotNull(messageChannelsMonitor); } @Test - public void testCircularReferenceNoChannelInFactoryBean() throws Exception { + public void testCircularReferenceNoChannelInFactoryBean() throws Exception { context = new GenericXmlApplicationContext(getClass(), "oref-factory-nonchannel.xml"); messageChannelsMonitor = context.getBean(IntegrationMBeanExporter.class); assertNotNull(messageChannelsMonitor); } @Test - public void testCircularReferenceWithChannel() throws Exception { + public void testCircularReferenceWithChannel() throws Exception { context = new GenericXmlApplicationContext(getClass(), "oref-channel.xml"); messageChannelsMonitor = context.getBean(IntegrationMBeanExporter.class); assertNotNull(messageChannelsMonitor); } @Test - public void testCircularReferenceWithChannelInFactoryBean() throws Exception { + public void testCircularReferenceWithChannelInFactoryBean() throws Exception { context = new GenericXmlApplicationContext(getClass(), "oref-factory-channel.xml"); messageChannelsMonitor = context.getBean(IntegrationMBeanExporter.class); assertNotNull(messageChannelsMonitor); @@ -87,14 +87,14 @@ public class MBeanExporterIntegrationTests { } @Test - public void testCircularReferenceWithChannelInFactoryBeanAutodetected() throws Exception { + public void testCircularReferenceWithChannelInFactoryBeanAutodetected() throws Exception { context = new GenericXmlApplicationContext(getClass(), "oref-factory-channel-autodetect.xml"); messageChannelsMonitor = context.getBean(IntegrationMBeanExporter.class); assertNotNull(messageChannelsMonitor); } @Test - public void testLifecycleInEndpointWithMessageSource() throws Exception { + public void testLifecycleInEndpointWithMessageSource() throws Exception { context = new GenericXmlApplicationContext(getClass(), "lifecycle-source.xml"); messageChannelsMonitor = context.getBean(IntegrationMBeanExporter.class); assertNotNull(messageChannelsMonitor); @@ -119,7 +119,8 @@ public class MBeanExporterIntegrationTests { ActiveChannel activeChannel = context.getBean("activeChannel", ActiveChannel.class); assertTrue(activeChannel.isStopCalled()); OtherActiveComponent otherActiveComponent = context.getBean(OtherActiveComponent.class); - assertTrue(otherActiveComponent.isStopCalled()); + assertTrue(otherActiveComponent.isBeforeCalled()); + assertTrue(otherActiveComponent.isAfterCalled()); } @Test @@ -138,7 +139,7 @@ public class MBeanExporterIntegrationTests { } @Test - public void testLifecycleInEndpointWithoutMessageSource() throws Exception { + public void testLifecycleInEndpointWithoutMessageSource() throws Exception { context = new GenericXmlApplicationContext(getClass(), "lifecycle-no-source.xml"); messageChannelsMonitor = context.getBean(IntegrationMBeanExporter.class); assertNotNull(messageChannelsMonitor); @@ -160,7 +161,7 @@ public class MBeanExporterIntegrationTests { } @Test - public void testComponentNames() throws Exception { + public void testComponentNames() throws Exception { context = new GenericXmlApplicationContext(getClass(), "excluded-components.xml"); messageChannelsMonitor = context.getBean(IntegrationMBeanExporter.class); assertNotNull(messageChannelsMonitor); @@ -173,7 +174,7 @@ public class MBeanExporterIntegrationTests { } @Test - public void testDuplicateComponentNames() throws Exception { + public void testDuplicateComponentNames() throws Exception { context = new GenericXmlApplicationContext(getClass(), "duplicate-components.xml"); messageChannelsMonitor = context.getBean(IntegrationMBeanExporter.class); assertNotNull(messageChannelsMonitor); @@ -181,12 +182,12 @@ public class MBeanExporterIntegrationTests { Set names = server.queryNames(ObjectName.getInstance("org.springframework.integration:type=ManagedEndpoint,*"), null); assertEquals(2, names.size()); } - + public static class BogusEndpoint extends AbstractEndpoint { - + @SuppressWarnings("unused") private IntegrationObjectSupport parent; - + public void setParent(IntegrationObjectSupport parent) { this.parent = parent; setComponentName(parent.getComponentName()); @@ -199,7 +200,7 @@ public class MBeanExporterIntegrationTests { @Override protected void doStop() { } - + } public static class DateFactoryBean implements FactoryBean { @@ -265,16 +266,16 @@ public class MBeanExporterIntegrationTests { } } - + @ManagedResource public static class Metric { - + } public static class MetricHolder implements InitializingBean { private MessageChannel channel; - + public void setChannel(MessageChannel channel) { this.channel = channel; } @@ -289,7 +290,7 @@ public class MBeanExporterIntegrationTests { String execute() throws Exception; int getCounter(); } - + public static class SimpleService implements Service { private int counter; @@ -336,23 +337,28 @@ public class MBeanExporterIntegrationTests { } } - public static class OtherActiveComponent implements OrderlyShutdownCapable, Lifecycle { + public static class OtherActiveComponent implements OrderlyShutdownCapable { - private boolean stopCalled; + private boolean beforeCalled; - public void start() { + private boolean afterCalled; + + public boolean isBeforeCalled() { + return this.beforeCalled; } - public void stop() { - this.stopCalled = true; + protected boolean isAfterCalled() { + return afterCalled; } - public boolean isRunning() { - return false; + public int beforeShutdown() { + this.beforeCalled = true; + return 0; } - public boolean isStopCalled() { - return this.stopCalled; + public int afterShutdown() { + this.afterCalled = true; + return 0; } } } diff --git a/src/reference/docbook/jmx.xml b/src/reference/docbook/jmx.xml index 6a560e9309..effe7f273b 100644 --- a/src/reference/docbook/jmx.xml +++ b/src/reference/docbook/jmx.xml @@ -421,6 +421,19 @@ seconds). +

+ Orderly Shutdown Managed Operation + + + The MBean exporter provides a JMX operation to shut down the application + in an orderly manner, intended for use before terminating the JVM. + + + + Its use and operation are described in . + +
diff --git a/src/reference/docbook/shutdown.xml b/src/reference/docbook/shutdown.xml new file mode 100644 index 0000000000..4ebeb1ecde --- /dev/null +++ b/src/reference/docbook/shutdown.xml @@ -0,0 +1,61 @@ + +
+ Orderly Shutdown + + + As described in , the MBean exporter provides a JMX operation + stopActiveComponents, which is used to stop the application in an orderly manner. The operation + has two parameters, a boolean and a long. The boolean indicates whether attempts will be made + to stop (interrupt) active threads; in most cases this will be set to false for orderly + shutdown. The long parameter indicates how long (in milliseconds) the operation will wait to allow + in-flight messages to complete. The operation works as follows: + + + The first step calls beforeShutdown() on all beans that implement + OrderlyShutdownCapable. This allows such components to prepare for shutdown. + Examples of components that implement this interface, and what they do with this call include: JMS and + AMQP message-driven adapters stop their listener containers; TCP server connection factories stop + accepting new connections (while keeping existing connections open); TCP inbound endpoints drop (log) + any new messages received; http inbound endpoints return 503 - Service Unavailable for any new + requests. + + + The second step stops any active channels, such as JMS- or AMQP-backed channels. + + + The third step stops all TaskSchedulers, preventing any new + scheduled operations (polling etc). + + + The fourth step stops all TaskExecutors, preventing any new + tasks from running. + + + If the shutdown is running from a Spring-managed TaskExecutor, shutting down that + executor would cause all the timeout time to be consumed by this step, because the thread won't terminate). + For this reason, either use a dedicated executor (via the shutdownExecutor property on the MBean exporter), + or do not use a Spring-managed executor to invoke this operation. + + + The fifth step stops all MessageSources. + + + The sixth step waits for any remaining time left, as defined by the value of the long parameter passed + in to the operation. This is intended to allow any in-flight messages to complete their journeys. It is + therefore important to select an appropriate timeout when invoking this operation. + + + The seventh step calls afterShutdown() on all OrderlyShutdownCapable components. + This allows such components to perform final shutdown tasks (closing all open sockets, for example). + + + If no time is left when we get to step 6, it probably means some thread is hung; in which case, the + operation attempts a forced shutdown on all schedulers and executors before exiting. + +
diff --git a/src/reference/docbook/system-management.xml b/src/reference/docbook/system-management.xml index e62904f29c..dacb6a1ea6 100644 --- a/src/reference/docbook/system-management.xml +++ b/src/reference/docbook/system-management.xml @@ -8,5 +8,6 @@ - + + diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index 4e22bfb60a..0063ad3d29 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -1,37 +1,37 @@ - What's new in Spring Integration 2.2? - - This chapter provides an overview of the new features and improvements - that have been introduced with Spring Integration 2.2 If you are interested - in even more detail, please take a look at the Issue Tracker tickets that - were resolved as part of the 2.2 development process: - + xmlns:xi="http://www.w3.org/2001/XInclude" + xmlns:xlink="http://www.w3.org/1999/xlink"> + What's new in Spring Integration 2.2? + + This chapter provides an overview of the new features and improvements + that have been introduced with Spring Integration 2.2 If you are interested + in even more detail, please take a look at the Issue Tracker tickets that + were resolved as part of the 2.2 development process: + -
- General -
- Spring 3.1 - - Spring Integration now uses Spring 3.1. - -
-
- Spring-AMQP 1.1 - - Spring Integration now uses Spring AMQP 1.1. This enables several features - to be used within a Spring Integration application, including... - - - A fixed reply queue for the outbound gateway - HA (mirrored) queues - Publisher Confirms - Returned Messages - Support for Dead Letter Exchanges/Dead Letter Queues - -
+
+ General +
+ Spring 3.1 + + Spring Integration now uses Spring 3.1. + +
+
+ Spring-AMQP 1.1 + + Spring Integration now uses Spring AMQP 1.1. This enables several features + to be used within a Spring Integration application, including... + + + A fixed reply queue for the outbound gateway + HA (mirrored) queues + Publisher Confirms + Returned Messages + Support for Dead Letter Exchanges/Dead Letter Queues + +
JDBC Adapter - Stored Procedures Components SpEL Support @@ -51,24 +51,34 @@ The Stored Procedure components now provide basic JMX support, exposing some of their properties as MBeans: - - Stored Procedure Name - Stored Procedure Name Expression - JdbcCallOperations Cache Statistics - -
-
- Transaction Synchronization - - When running from a transactional poller, - mail inbound adapters can be configured to update the mailbox only - if the transaction commits. - -
-
+ + Stored Procedure Name + Stored Procedure Name Expression + JdbcCallOperations Cache Statistics + +
+
+ Transaction Synchronization + + When running from a transactional poller, + mail inbound adapters can be configured to update the mailbox only + if the transaction commits. + +
+
+ Orderly Shutdown + + A method stopActiveComponents() has been + added to the IntegrationMBeanExporter. This allows a Spring Integration + application to be shut down in an orderly manner, disallowing new inbound + messages to certain adapters and waiting for some time to allow in-flight + messages to complete. + +
+ -
- New Components +
+ New Components
JPA Endpoints @@ -94,9 +104,9 @@ For more information please see
-
-
- Framework Refactoring +
+
+ Framework Refactoring -
+