diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java index 3fa43c4902..58191bc1a3 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java @@ -60,6 +60,7 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageDeliveryException; import org.springframework.messaging.MessageHandlingException; +import org.springframework.scheduling.TaskScheduler; import org.springframework.util.Assert; import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; @@ -144,6 +145,12 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp private volatile boolean requiresReply; + private long lastSend; + + private volatile long idleReplyContainerTimeout; + + private ScheduledFuture idleTask; + /** * Set whether message delivery should be persistent or non-persistent, * specified as a boolean value ("true" or "false"). This will set the delivery @@ -425,6 +432,29 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp this.requiresReply = requiresReply; } + /** + * Set the target timeout for idle containers, in seconds. Setting this greater than zero enables lazy + * starting of the reply listener container. The container will be started when a message is sent. It will be + * stopped when idle for at least this time. The actual stop time may be up to 1.5x this time. + * @param idleReplyContainerTimeout the timeout in seconds. + * @since 4.2 + */ + public void setIdleReplyContainerTimeout(long idleReplyContainerTimeout) { + setIdleReplyContainerTimeout(idleReplyContainerTimeout, TimeUnit.SECONDS); + } + + /** + * Set the target timeout for idle containers. Setting this greater than zero enables lazy + * starting of the reply listener container. The container will be started when a message is sent. It will be + * stopped when idle for at least this time. The actual stop time may be up to 1.5x this time. + * @param idleReplyContainerTimeout the timeout in seconds. + * @param unit the time unit. + * @since 4.2 + */ + public void setIdleReplyContainerTimeout(long idleReplyContainerTimeout, TimeUnit unit) { + this.idleReplyContainerTimeout = unit.toMillis(idleReplyContainerTimeout); + } + private Destination determineRequestDestination(Message message, Session session) throws JMSException { if (this.requestDestination != null) { return this.requestDestination; @@ -620,9 +650,16 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp synchronized (this.lifeCycleMonitor) { if (!this.active) { if (this.replyContainer != null) { - this.replyContainer.start(); + TaskScheduler taskScheduler = getTaskScheduler(); + if (this.idleReplyContainerTimeout <= 0) { + this.replyContainer.start(); + } + else { + Assert.state(taskScheduler != null, "'taskScheduler' is required."); + } if (this.receiveTimeout >= 0) { - this.reaper = this.getTaskScheduler().schedule(new LateReplyReaper(), new Date()); + Assert.state(taskScheduler != null, "'taskScheduler' is required."); + this.reaper = taskScheduler.schedule(new LateReplyReaper(), new Date()); } } this.active = true; @@ -638,6 +675,10 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp this.deleteDestinationIfTemporary(this.replyContainer.getDestination()); this.reaper.cancel(false); } + if (this.idleTask != null) { + this.idleTask.cancel(true); + this.idleTask = null; + } this.active = false; } } @@ -659,6 +700,19 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp jmsReply = this.sendAndReceiveWithoutContainer(requestMessage); } else { + if (this.idleReplyContainerTimeout > 0) { + synchronized(this.lifeCycleMonitor) { + this.lastSend = System.currentTimeMillis(); + if (!this.replyContainer.isRunning()) { + if (logger.isDebugEnabled()) { + logger.debug(this.getComponentName() + ": Starting reply container."); + } + this.replyContainer.start(); + this.idleTask = getTaskScheduler().scheduleAtFixedRate(new IdleContainerStopper(), + this.idleReplyContainerTimeout / 2); + } + } + } jmsReply = this.sendAndReceiveWithContainer(requestMessage); } if (jmsReply == null) { @@ -1275,7 +1329,29 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp } } + private class IdleContainerStopper implements Runnable { + + @Override + public void run() { + synchronized(JmsOutboundGateway.this.lifeCycleMonitor) { + if (System.currentTimeMillis() - lastSend > idleReplyContainerTimeout + && replies.size() == 0) { + if (replyContainer.isRunning()) { + if (logger.isDebugEnabled()) { + logger.debug(getComponentName() + ": Stopping idle reply container."); + } + replyContainer.stop(); + idleTask.cancel(false); + idleTask = null; + } + } + } + } + + } + public static class ReplyContainerProperties { + private volatile Boolean sessionTransacted; private volatile Integer sessionAcknowledgeMode; diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsOutboundGatewayParser.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsOutboundGatewayParser.java index ca76d5ff5f..b45eeaf3e7 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsOutboundGatewayParser.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsOutboundGatewayParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2015 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. @@ -16,6 +16,8 @@ package org.springframework.integration.jms.config; +import org.w3c.dom.Element; + import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.config.ExpressionFactoryBean; @@ -24,7 +26,6 @@ import org.springframework.integration.config.xml.IntegrationNamespaceUtils; import org.springframework.integration.jms.JmsOutboundGateway; import org.springframework.util.StringUtils; import org.springframework.util.xml.DomUtils; -import org.w3c.dom.Element; /** * Parser for the <outbound-gateway> element of the integration 'jms' namespace. @@ -66,13 +67,15 @@ public class JmsOutboundGatewayParser extends AbstractConsumerEndpointParser { IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "priority"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "explicit-qos-enabled"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "requires-reply"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "idle-reply-listener-timeout", + "idleReplyContainerTimeout"); String deliveryPersistent = element.getAttribute("delivery-persistent"); if (StringUtils.hasText(deliveryPersistent)) { builder.addPropertyValue("deliveryPersistent", deliveryPersistent); } Element container = DomUtils.getChildElementByTagName(element, "reply-listener"); - + if (container != null) { this.parseReplyContainer(builder, parserContext, container); } diff --git a/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-4.2.xsd b/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-4.2.xsd index 51e14de363..7f20ced2d5 100644 --- a/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-4.2.xsd +++ b/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-4.2.xsd @@ -1108,6 +1108,15 @@ + + + + When using a 'reply-listener', specify whether the container should be started on-demand + and stopped when idle for this time (in seconds). When omitted (or <= 0), the reply container + is started/stopped according to the gateway's lifecycle. + + + diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayFunctionTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayFunctionTests.java index 55c34060fe..61bc04e52e 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayFunctionTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayFunctionTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2015 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. @@ -15,6 +15,7 @@ */ package org.springframework.integration.jms; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; import static org.mockito.Mockito.mock; @@ -34,12 +35,16 @@ import javax.jms.Session; import org.apache.activemq.ActiveMQConnectionFactory; import org.apache.activemq.command.ActiveMQQueue; import org.junit.Test; + import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.context.IntegrationContextUtils; -import org.springframework.messaging.support.GenericMessage; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.jms.JmsException; import org.springframework.jms.connection.CachingConnectionFactory; import org.springframework.jms.core.JmsTemplate; import org.springframework.jms.core.MessageCreator; +import org.springframework.jms.listener.DefaultMessageListenerContainer; +import org.springframework.messaging.support.GenericMessage; import org.springframework.scheduling.TaskScheduler; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; @@ -60,19 +65,15 @@ public class OutboundGatewayFunctionTests { private static Destination requestQueue3 = new ActiveMQQueue("request3"); - private static Destination replyQueue3 = new ActiveMQQueue("reply3"); - private static Destination requestQueue4 = new ActiveMQQueue("request4"); - private static Destination replyQueue4 = new ActiveMQQueue("reply4"); - private static Destination requestQueue5 = new ActiveMQQueue("request5"); - private static Destination replyQueue5 = new ActiveMQQueue("reply5"); - private static Destination requestQueue6 = new ActiveMQQueue("request6"); - private static Destination replyQueue6 = new ActiveMQQueue("reply6"); + private static Destination requestQueue7 = new ActiveMQQueue("request7"); + + private static Destination replyQueue7 = new ActiveMQQueue("reply7"); @Test public void testContainerWithDest() throws Exception { @@ -95,6 +96,7 @@ public class OutboundGatewayFunctionTests { final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { latch1.countDown(); try { @@ -114,6 +116,7 @@ public class OutboundGatewayFunctionTests { final javax.jms.Message jmsReply = request; template.send(request.getJMSReplyTo(), new MessageCreator() { + @Override public Message createMessage(Session session) throws JMSException { return jmsReply; } @@ -144,6 +147,7 @@ public class OutboundGatewayFunctionTests { final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { latch1.countDown(); try { @@ -163,6 +167,7 @@ public class OutboundGatewayFunctionTests { final javax.jms.Message jmsReply = request; template.send(request.getJMSReplyTo(), new MessageCreator() { + @Override public Message createMessage(Session session) throws JMSException { jmsReply.setJMSCorrelationID(jmsReply.getJMSMessageID()); return jmsReply; @@ -195,6 +200,7 @@ public class OutboundGatewayFunctionTests { final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { latch1.countDown(); try { @@ -214,6 +220,7 @@ public class OutboundGatewayFunctionTests { final javax.jms.Message jmsReply = request; template.send(request.getJMSReplyTo(), new MessageCreator() { + @Override public Message createMessage(Session session) throws JMSException { return jmsReply; } @@ -244,6 +251,7 @@ public class OutboundGatewayFunctionTests { final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { latch1.countDown(); try { @@ -263,6 +271,7 @@ public class OutboundGatewayFunctionTests { final javax.jms.Message jmsReply = request; template.send(request.getJMSReplyTo(), new MessageCreator() { + @Override public Message createMessage(Session session) throws JMSException { jmsReply.setJMSCorrelationID(jmsReply.getJMSMessageID()); return jmsReply; @@ -294,6 +303,7 @@ public class OutboundGatewayFunctionTests { final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { latch1.countDown(); try { @@ -313,6 +323,7 @@ public class OutboundGatewayFunctionTests { final javax.jms.Message jmsReply = request; template.send(request.getJMSReplyTo(), new MessageCreator() { + @Override public Message createMessage(Session session) throws JMSException { return jmsReply; } @@ -342,6 +353,7 @@ public class OutboundGatewayFunctionTests { final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { latch1.countDown(); try { @@ -361,6 +373,7 @@ public class OutboundGatewayFunctionTests { final javax.jms.Message jmsReply = request; template.send(request.getJMSReplyTo(), new MessageCreator() { + @Override public Message createMessage(Session session) throws JMSException { jmsReply.setJMSCorrelationID(jmsReply.getJMSMessageID()); return jmsReply; @@ -372,6 +385,68 @@ public class OutboundGatewayFunctionTests { gateway.stop(); } + @Test + public void testLazyContainerWithDest() throws Exception { + BeanFactory beanFactory = mock(BeanFactory.class); + when(beanFactory.containsBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME)).thenReturn(true); + ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); + scheduler.initialize(); + when(beanFactory.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class)) + .thenReturn(scheduler); + final JmsOutboundGateway gateway = new JmsOutboundGateway(); + gateway.setBeanFactory(beanFactory); + gateway.setConnectionFactory(getGatewayConnectionFactory()); + gateway.setRequestDestination(requestQueue7); + gateway.setReplyDestination(replyQueue7); + gateway.setCorrelationKey("JMSCorrelationID"); + gateway.setUseReplyContainer(true); + gateway.setIdleReplyContainerTimeout(1, TimeUnit.SECONDS); + gateway.afterPropertiesSet(); + gateway.start(); + Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override + public void run() { + JmsTemplate template = new JmsTemplate(); + template.setConnectionFactory(getTemplateConnectionFactory()); + template.setReceiveTimeout(10000); + receiveAndSend(template); + receiveAndSend(template); + } + + private void receiveAndSend(JmsTemplate template) { + javax.jms.Message request = template.receive(requestQueue7); + final javax.jms.Message jmsReply = request; + try { + template.send(request.getJMSReplyTo(), new MessageCreator() { + + @Override + public Message createMessage(Session session) throws JMSException { + return jmsReply; + } + }); + } + catch (JmsException e) { + } + catch (JMSException e) { + } + } + }); + + assertNotNull(gateway.handleRequestMessage(new GenericMessage("foo"))); + DefaultMessageListenerContainer container = TestUtils.getPropertyValue(gateway, "replyContainer", + DefaultMessageListenerContainer.class); + int n = 0; + while (n++ < 100 && container.isRunning()) { + Thread.sleep(100); + } + assertFalse(container.isRunning()); + assertNotNull(gateway.handleRequestMessage(new GenericMessage("foo"))); + assertTrue(container.isRunning()); + + gateway.stop(); + assertFalse(container.isRunning()); + } + private ConnectionFactory getTemplateConnectionFactory() { ConnectionFactory amqConnectionFactory = new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false"); return amqConnectionFactory; diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsOutboundGatewayParserTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsOutboundGatewayParserTests.java index fe1276b00a..ac37df71e7 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsOutboundGatewayParserTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsOutboundGatewayParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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. @@ -16,8 +16,17 @@ package org.springframework.integration.jms.config; -import static org.junit.Assert.*; -import static org.mockito.Mockito.*; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; import java.lang.reflect.Method; import java.util.Properties; @@ -94,6 +103,8 @@ public class JmsOutboundGatewayParserTests { assertEquals(3, TestUtils.getPropertyValue(container, "cacheLevel")); assertTrue(container.isSessionTransacted()); assertSame(context.getBean("exec"), TestUtils.getPropertyValue(container, "taskExecutor")); + assertEquals(1234000L, TestUtils.getPropertyValue(gateway, "idleReplyContainerTimeout")); + context.close(); } @Test @@ -105,6 +116,7 @@ public class JmsOutboundGatewayParserTests { gateway.handleMessage(new GenericMessage("foo")); assertEquals(1, adviceCalled); assertEquals(3, TestUtils.getPropertyValue(gateway, "replyContainer.sessionAcknowledgeMode")); + context.close(); } @Test @@ -117,6 +129,7 @@ public class JmsOutboundGatewayParserTests { accessor = new DirectFieldAccessor(gateway); MessageConverter converter = (MessageConverter)accessor.getPropertyValue("messageConverter"); assertTrue("Wrong message converter", converter instanceof StubMessageConverter); + context.close(); } @Test @@ -129,6 +142,7 @@ public class JmsOutboundGatewayParserTests { Object order = accessor.getPropertyValue("order"); assertEquals(99, order); assertEquals(Boolean.TRUE, accessor.getPropertyValue("requiresReply")); + context.close(); } @Test @@ -140,6 +154,7 @@ public class JmsOutboundGatewayParserTests { JmsOutboundGateway gateway = (JmsOutboundGateway) accessor.getPropertyValue("handler"); accessor = new DirectFieldAccessor(gateway); assertSame(context.getBean("replyQueue"), accessor.getPropertyValue("replyDestination")); + context.close(); } @Test @@ -151,6 +166,7 @@ public class JmsOutboundGatewayParserTests { JmsOutboundGateway gateway = (JmsOutboundGateway) accessor.getPropertyValue("handler"); accessor = new DirectFieldAccessor(gateway); assertEquals("replyQueueName", accessor.getPropertyValue("replyDestinationName")); + context.close(); } @Test @@ -176,6 +192,7 @@ public class JmsOutboundGatewayParserTests { when(session.createQueue("foo")).thenReturn(queue); Destination replyQ = (Destination) method.invoke(gateway, message, session); assertSame(queue, replyQ); + context.close(); } @Test @@ -191,13 +208,14 @@ public class JmsOutboundGatewayParserTests { Expression.class); assertEquals("@replyQueue", expression.getExpressionString()); assertSame(context.getBean("replyQueue"), processor.processMessage(null)); + context.close(); } @Test public void gatewayWithDestAndDestExpression() { try { new ClassPathXmlApplicationContext( - "jmsOutboundGatewayReplyDestOptions-fail.xml", this.getClass()); + "jmsOutboundGatewayReplyDestOptions-fail.xml", this.getClass()).close(); fail("Exception expected"); } catch (BeanDefinitionParsingException e) { @@ -214,15 +232,16 @@ public class JmsOutboundGatewayParserTests { SampleGateway gateway = context.getBean("gateway", SampleGateway.class); SubscribableChannel jmsInput = context.getBean("jmsInput", SubscribableChannel.class); MessageHandler handler = new MessageHandler() { + @Override public void handleMessage(Message message) throws MessagingException { MessageHistory history = MessageHistory.read(message); assertNotNull(history); Properties componentHistoryRecord = TestUtils.locateComponentInHistory(history, "inboundGateway", 0); assertNotNull(componentHistoryRecord); assertEquals("jms:inbound-gateway", componentHistoryRecord.get("type")); - MessagingTemplate messagingTemplate = new MessagingTemplate(); - messagingTemplate.setDefaultDestination((MessageChannel)message.getHeaders().getReplyChannel()); - messagingTemplate.send(message); + MessagingTemplate messagingTemplate = new MessagingTemplate(); + messagingTemplate.setDefaultDestination((MessageChannel)message.getHeaders().getReplyChannel()); + messagingTemplate.send(message); } }; handler = spy(handler); @@ -230,6 +249,7 @@ public class JmsOutboundGatewayParserTests { String result = gateway.echo("hello"); verify(handler, times(1)).handleMessage(Mockito.any(Message.class)); assertEquals("hello", result); + context.close(); } @Test @@ -241,6 +261,7 @@ public class JmsOutboundGatewayParserTests { new DirectFieldAccessor(endpoint).getPropertyValue("handler")); assertFalse((Boolean) accessor.getPropertyValue("requestPubSubDomain")); assertFalse((Boolean) accessor.getPropertyValue("replyPubSubDomain")); + context.close(); } @Test @@ -252,6 +273,7 @@ public class JmsOutboundGatewayParserTests { new DirectFieldAccessor(endpoint).getPropertyValue("handler")); assertTrue((Boolean) accessor.getPropertyValue("requestPubSubDomain")); assertTrue((Boolean) accessor.getPropertyValue("replyPubSubDomain")); + context.close(); } diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/jmsOutboundGatewayWithDeliveryPersistent.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/jmsOutboundGatewayWithDeliveryPersistent.xml index 57c94fd370..ef85b04198 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/jmsOutboundGatewayWithDeliveryPersistent.xml +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/jmsOutboundGatewayWithDeliveryPersistent.xml @@ -30,6 +30,7 @@ request-destination-name="requestQueue" request-channel="requestChannel" delivery-persistent="true" + idle-reply-listener-timeout="1234" auto-startup="false"> + request-destination="outQueue" + request-channel="outboundJmsRequests" + reply-channel="jmsReplies"> ---- @@ -266,6 +266,20 @@ The listener is very lightweight and it is anticipated that, in most cases, only However, attributes such as _concurrent-consumers_, _max-concurrent-consumers_ etc., can be added. Refer to the schema for a complete list of supported attributes, together with thehttp://static.springsource.org/spring/docs/current/spring-framework-reference/html/jms.html[Spring JMS documentation] for their meanings. +*Idle Reply Listeners* + +Starting with _version 4.2_, the reply listener can be started as needed (and stopped after an idle time) instead +of running for the duration of the gateway's lifecycle. +This might be useful if you have many gateways in the application context where they are mostly idle. +One such situation is a context with many (inactive) partitioned http://projects.spring.io/spring-batch/[Spring Batch] +jobs using Spring Integration and JMS for partition distribution. +If all the reply listeners were active, the JMS broker would have an active consumer for each gateway. +By enabling the idle timeout, each consumer would exist only while the corresponding batch job is running (and +for a short time after it finishes). + +See `idle-reply-listener-timeout` in <>. + +[[jms-og-attributes]] ==== Attribute Reference [source,xml] @@ -295,7 +309,8 @@ Refer to the schema for a complete list of supported attributes, together with t request-pub-sub-domain="" <22> time-to-live="" <23> requires-reply=""> <24> - <25> + idle-reply-listener-timeout <25> + <26> ---- @@ -401,8 +416,15 @@ This value is `true` by default, and a `MessageTimeoutException` will be thrown Note, it is important to keep in mind that, if the service is never expected to return a reply, it would be better to use a `` instead of a `` with `requires-reply="false"`. With the latter, the sending thread is blocked, waiting for a reply for the `receive-timeout` period. +<25> When a `` is used, it's lifecycle (start/stop) matches that of the gateway by default. +When this value is greater than `0`, the container is started on demand (when a request is sent). +The container continues to run until at least this time elapses with no requests being received (and no replies +are outstanding). +The container will be started again on the next request. +The stop time is a minimum and may actually be up to 1.5x this value. -<25> When this element is included, replies are received by a `MessageListenerContainer` rather than creating a consumer for each reply. +<26> When this element is included, replies are received by an asynchronous `MessageListenerContainer` rather than +creating a consumer for each reply. This can be more efficient in many cases. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index b55cbb9afd..070671b24f 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -155,6 +155,13 @@ Of course, Reactor is still supported for functionality such as the `Promise` ga [[x4.2-jms-changes]] ==== JMS Changes +===== Reply Listener Lazy Initialization + +It is now possible to configure the reply listener in JMS outbound gateways to be initialized on-demand and stopped +after an idle period, instead of being controlled by the gateway's lifecycle. + +See <> for more information. + ===== Conversion Errors in Message-Driven Endpoints The `error-channel` now is used for the conversion errors, which have caused a transaction rollback and message redelivery previously.