INT-3005: Add throwExceptionOnLateReply Property

JIRA: https://jira.springsource.org/browse/INT-3005

* `GenericMessagingTemplate` has `throwExceptionOnLateReply` property.
This change provides a global `messagingTemplate.throwExceptionOnLateReply` property
to allow change default `false` to `true` using `spring.integration.properties`

* Refactor `spring.integration.properties` keys to be consistent with other Spring properties - adde prefix `spring.integraton.`

INT-3005: Add test for `lateReply`

INT-3279: Fix classes tangle

spring-integration-core/src/main/java/org/springframework/integration/aop/MessagePublishingInterceptor.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2013 the original author or authors.
+ * Copyright 2002-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.
@@ -51,6 +51,7 @@ import org.springframework.util.StringUtils;
  * responsibility of the EL expression provided by the {@link PublisherMetadataSource}.
  *
  * @author Mark Fisher
+ * @author Artem Bilan
  * @since 2.0
  */
 public class MessagePublishingInterceptor implements MethodInterceptor, BeanFactoryAware {
@@ -90,6 +91,7 @@ public class MessagePublishingInterceptor implements MethodInterceptor, BeanFact
 	@Override
 	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
 		this.beanFactory = beanFactory;
+		this.messagingTemplate.setBeanFactory(beanFactory);
 	}
 
 	public final Object invoke(final MethodInvocation invocation) throws Throwable {

spring-integration-core/src/main/java/org/springframework/integration/config/xml/DefaultConfiguringBeanFactoryPostProcessor.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2013 the original author or authors.
+ * Copyright 2002-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.
@@ -154,7 +154,7 @@ class DefaultConfiguringBeanFactoryPostProcessor implements BeanFactoryPostProce
 		}
 		BeanDefinitionBuilder schedulerBuilder = BeanDefinitionBuilder.genericBeanDefinition(
 				"org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler");
-		String taskSchedulerPoolSizeExpression = IntegrationProperties.getExpressionFor(IntegrationProperties.TASKSCHEDULER_POOLSIZE);
+		String taskSchedulerPoolSizeExpression = IntegrationProperties.getExpressionFor(IntegrationProperties.TASK_SCHEDULER_POOL_SIZE);
 		schedulerBuilder.addPropertyValue("poolSize", taskSchedulerPoolSizeExpression);
 		schedulerBuilder.addPropertyValue("threadNamePrefix", "task-scheduler-");
 		schedulerBuilder.addPropertyValue("rejectedExecutionHandler", new CallerRunsPolicy());

spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationProperties.java

@@ -33,33 +33,38 @@ import org.springframework.core.io.support.ResourcePatternResolver;
  */
 public final class IntegrationProperties {
 
+	public static final String INTEGRATION_PROPERTIES_PREFIX = "spring.integraton.";
+
 	/**
 	 * Specifies whether to allow create automatically {@link org.springframework.integration.channel.DirectChannel}
 	 * beans for non-declared channels or not.
 	 */
-	public static final String CHANNELS_AUTOCREATE = "channels.autoCreate";
+	public static final String CHANNELS_AUTOCREATE = INTEGRATION_PROPERTIES_PREFIX + "channels.autoCreate";
 
 	/**
 	 * Specifies the value for {@link org.springframework.integration.dispatcher.UnicastingDispatcher#maxSubscribers}
 	 * in case of point-to-point channels (e.g. {@link org.springframework.integration.channel.ExecutorChannel}),
 	 * if the attribute {@code max-subscribers} isn't configured on the channel component.
 	 */
-	public static final String CHANNELS_MAX_UNICAST_SUBSCRIBERS = "channels.maxUnicastSubscribers";
+	public static final String CHANNELS_MAX_UNICAST_SUBSCRIBERS = INTEGRATION_PROPERTIES_PREFIX + "channels.maxUnicastSubscribers";
 
 	/**
 	 * Specifies the value for {@link org.springframework.integration.dispatcher.BroadcastingDispatcher#maxSubscribers}
 	 * in case of point-to-point channels (e.g. {@link org.springframework.integration.channel.PublishSubscribeChannel}),
 	 * if the attribute {@code max-subscribers} isn't configured on the channel component.
 	 */
-	public static final String CHANNELS_MAX_BROADCAST_SUBSCRIBERS = "channels.maxBroadcastSubscribers";
+	public static final String CHANNELS_MAX_BROADCAST_SUBSCRIBERS = INTEGRATION_PROPERTIES_PREFIX + "channels.maxBroadcastSubscribers";
 
 	/**
 	 * Specifies the value of {@link org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler#poolSize}
-	 * for the {@code taskScheduler} bean initialized by the AbstractTransformerIntegration infrastructure.
+	 * for the {@code taskScheduler} bean initialized by the Integration infrastructure.
 	 */
-	public static final String TASKSCHEDULER_POOLSIZE = "taskScheduler.poolSize";
+	public static final String TASK_SCHEDULER_POOL_SIZE = INTEGRATION_PROPERTIES_PREFIX + "taskScheduler.poolSize";
 
-//	TODO public static final String LATE_REPLY_LOGGING_LEVEL = "messagingTemplate.lateReply.logging.level";
+	/**
+	 * Specifies the value of {@link org.springframework.messaging.core.GenericMessagingTemplate#throwExceptionOnLateReply}.
+	 */
+	public static final String THROW_EXCEPTION_ON_LATE_REPLY = INTEGRATION_PROPERTIES_PREFIX + "messagingTemplate.throwExceptionOnLateReply";
 
 	private static Properties defaults;
 
@@ -97,7 +102,7 @@ public final class IntegrationProperties {
 	 */
 	public static String getExpressionFor(String key) {
 		if (defaults.containsKey(key)) {
-			return "#{T(" + IntegrationContextUtils.class.getName() + ").getIntegrationProperties(beanFactory).getProperty('" + key + "')}";
+			return "#{T(org.springframework.integration.context.IntegrationContextUtils).getIntegrationProperties(beanFactory).getProperty('" + key + "')}";
 		}
 		else {
 			throw new IllegalArgumentException("The provided key [" + key + "] isn't the one of Integration properties: " + defaults.keySet());

spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013 the original author or authors.
+ * Copyright 2013-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.
@@ -15,14 +15,19 @@
  */
 package org.springframework.integration.core;
 
+import java.util.Properties;
+
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactory;
+import org.springframework.integration.context.IntegrationContextUtils;
+import org.springframework.integration.context.IntegrationProperties;
 import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.core.GenericMessagingTemplate;
 
 /**
  * @author Gary Russell
+ * @author Artem Bilan
  * @since 3.0
  *
  */
@@ -34,6 +39,9 @@ public class MessagingTemplate extends GenericMessagingTemplate {
 	@Override
 	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
 		super.setDestinationResolver(new BeanFactoryChannelResolver(beanFactory));
+		Properties integrationProperties = IntegrationContextUtils.getIntegrationProperties(beanFactory);
+		Boolean throwExceptionOnLateReply = Boolean.valueOf(integrationProperties.getProperty(IntegrationProperties.THROW_EXCEPTION_ON_LATE_REPLY));
+		this.setThrowExceptionOnLateReply(throwExceptionOnLateReply);
 	}
 
 	/**

spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-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.
@@ -32,6 +32,7 @@ import org.springframework.util.Assert;
  * output channel and a convenience method for sending Messages.
  *
  * @author Mark Fisher
+ * @author Artem Bilan
  */
 public abstract class MessageProducerSupport extends AbstractEndpoint implements MessageProducer, TrackableComponent {
 
@@ -63,6 +64,9 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements
 	@Override
 	protected void onInit() {
 		Assert.notNull(this.outputChannel, "outputChannel is required");
+		if (this.getBeanFactory() != null) {
+			this.messagingTemplate.setBeanFactory(this.getBeanFactory());
+		}
 	}
 
 	/**

spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java

@@ -34,6 +34,7 @@ import org.springframework.util.Assert;
  * @author Mark Fisher
  * @author Oleg Zhurakousky
  * @author Gary Russell
+ * @author Artem Bilan
  */
 public class SourcePollingChannelAdapter extends AbstractPollingEndpoint
 		implements TrackableComponent {
@@ -95,6 +96,9 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint
 		Assert.notNull(this.source, "source must not be null");
 		Assert.notNull(this.outputChannel, "outputChannel must not be null");
 		super.onInit();
+		if (this.getBeanFactory() != null) {
+			this.messagingTemplate.setBeanFactory(this.getBeanFactory());
+		}
 	}
 
 	@Override

spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java

@@ -43,6 +43,7 @@ import org.springframework.util.Assert;
  *
  * @author Mark Fisher
  * @author Gary Russell
+ * @author Artem Bilan
  */
 public abstract class MessagingGatewaySupport extends AbstractEndpoint implements TrackableComponent {
 
@@ -172,6 +173,9 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement
 	@Override
 	protected void onInit() throws Exception {
 		this.historyWritingPostProcessor.setTrackableComponent(this);
+		if (this.getBeanFactory() != null) {
+			this.messagingTemplate.setBeanFactory(this.getBeanFactory());
+		}
 		this.initialized = true;
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/handler/advice/ErrorMessageSendingRecoverer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2013 the original author or authors.
+ * Copyright 2002-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.
@@ -18,6 +18,9 @@ package org.springframework.integration.handler.advice;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.BeanFactoryAware;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessagingException;
@@ -36,12 +39,14 @@ import org.springframework.util.Assert;
  * @since 2.2
  *
  */
-public class ErrorMessageSendingRecoverer implements RecoveryCallback<Object> {
+public class ErrorMessageSendingRecoverer implements RecoveryCallback<Object>, BeanFactoryAware {
 
 	private final static Log logger = LogFactory.getLog(ErrorMessageSendingRecoverer.class);
 
 	private final MessagingTemplate messagingTemplate = new MessagingTemplate();
 
+	private BeanFactory beanFactory;
+
 	public ErrorMessageSendingRecoverer(MessageChannel channel) {
 		Assert.notNull(channel, "channel cannot be null");
 		this.messagingTemplate.setDefaultDestination(channel);
@@ -51,6 +56,12 @@ public class ErrorMessageSendingRecoverer implements RecoveryCallback<Object> {
 		this.messagingTemplate.setSendTimeout(sendTimeout);
 	}
 
+	@Override
+	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
+		this.beanFactory = beanFactory;
+		this.messagingTemplate.setBeanFactory(beanFactory);
+	}
+
 	public Object recover(RetryContext context) throws Exception {
 		Throwable lastThrowable = context.getLastThrowable();
 		if (lastThrowable == null) {

spring-integration-core/src/main/java/org/springframework/integration/handler/advice/ExpressionEvaluatingRequestHandlerAdvice.java

@@ -36,7 +36,9 @@ import org.springframework.util.Assert;
  * or onFailureChannel as appropriate; the message is the input message with a header
  * {@link org.springframework.integration.IntegrationMessageHeaderAccessor#POSTPROCESS_RESULT} containing the evaluation result.
  * The failure expression is NOT evaluated if the success expression throws an exception.
+ *
  * @author Gary Russell
+ * @author Artem Bilan
  * @since 2.2
  *
  */
@@ -109,6 +111,14 @@ public class ExpressionEvaluatingRequestHandlerAdvice extends AbstractRequestHan
 		this.propagateOnSuccessEvaluationFailures = propagateOnSuccessEvaluationFailures;
 	}
 
+	@Override
+	protected void onInit() throws Exception {
+		super.onInit();
+		if (this.getBeanFactory() != null) {
+			this.messagingTemplate.setBeanFactory(this.getBeanFactory());
+		}
+	}
+
 	@Override
 	protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
 		try {

spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java

@@ -38,6 +38,7 @@ import org.springframework.messaging.MessagingException;
  * @author Gunnar Hillert
  * @author Soby Chacko
  * @author Stefan Ferstl
+ * @author Artem Bilan
  */
 @ManagedResource
 public abstract class AbstractMessageRouter extends AbstractMessageHandler {
@@ -123,6 +124,14 @@ public abstract class AbstractMessageRouter extends AbstractMessageHandler {
 		return this.getConversionService();
 	}
 
+	@Override
+	protected void onInit() throws Exception {
+		super.onInit();
+		if (this.getBeanFactory() != null) {
+			this.messagingTemplate.setBeanFactory(this.getBeanFactory());
+		}
+	}
+
 	/**
 	 * Subclasses must implement this method to return a Collection of zero or more
 	 * MessageChannels to which the given Message should be routed.

spring-integration-core/src/main/resources/META-INF/spring.integration.default.properties

@@ -1,4 +1,5 @@
-channels.autoCreate=true
-channels.maxUnicastSubscribers=0x7fffffff
-channels.maxBroadcastSubscribers=0x7fffffff
-taskScheduler.poolSize=10
+spring.integraton.channels.autoCreate=true
+spring.integraton.channels.maxUnicastSubscribers=0x7fffffff
+spring.integraton.channels.maxBroadcastSubscribers=0x7fffffff
+spring.integraton.taskScheduler.poolSize=10
+spring.integraton.messagingTemplate.throwExceptionOnLateReply=false

spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersOverrideDefaultTests-context.xml

@@ -8,8 +8,8 @@
 		http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd">
 
 	<util:properties id="integrationGlobalProperties">
-		<prop key="channels.maxUnicastSubscribers">456</prop>
-		<prop key="channels.maxBroadcastSubscribers">789</prop>
+		<prop key="spring.integraton.channels.maxUnicastSubscribers">456</prop>
+		<prop key="spring.integraton.channels.maxBroadcastSubscribers">789</prop>
 	</util:properties>
 
 	<import resource="DispatcherMaxSubscribersDefaultConfigurationTests-context.xml" />

spring-integration-core/src/test/java/org/springframework/integration/context/IntegrationContextTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013 the original author or authors.
+ * Copyright 2013-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.
@@ -51,8 +51,8 @@ public class IntegrationContextTests {
 
 	@Test
 	public void testIntegrationContextComponents() {
-		//TODO INT-3005  assertEquals("error", this.integrationProperties.get(IntegrationProperties.LATE_REPLY_LOGGING_LEVEL));
-		assertEquals("20", this.integrationProperties.get(IntegrationProperties.TASKSCHEDULER_POOLSIZE));
+		assertEquals("true", this.integrationProperties.get(IntegrationProperties.THROW_EXCEPTION_ON_LATE_REPLY));
+		assertEquals("20", this.integrationProperties.get(IntegrationProperties.TASK_SCHEDULER_POOL_SIZE));
 		assertEquals(this.integrationProperties, this.serviceActivator.getIntegrationProperties());
 		assertEquals(20, TestUtils.getPropertyValue(this.taskScheduler, "poolSize"));
 	}

spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests-context.xml

@@ -5,17 +5,31 @@
 	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
 		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
 
+	<int:channel id="errorChannel">
+		<int:queue/>
+	</int:channel>
+
 	<int:gateway id="sampleGateway"
 			service-interface="org.springframework.integration.gateway.GatewayInterfaceTests.Bar"
-			default-request-channel="requestChannelBaz">
+			default-request-channel="requestChannelBaz"
+			error-channel="errorChannel">
 		<int:default-header name="name" expression="#gatewayMethod.name"/>
 		<int:default-header name="string" expression="#gatewayMethod.toString()"/>
 		<int:default-header name="object" expression="#gatewayMethod"/>
 		<int:method name="baz">
 			<int:header name="name" value="overrideGlobal"/>
 		</int:method>
+		<int:method name="lateReply" request-channel="lateReplyChannel" reply-timeout="0"/>
 	</int:gateway>
 
+	<int:chain input-channel="lateReplyChannel" >
+		<int:header-enricher>
+			<int:error-channel ref="errorChannel" overwrite="true"/>
+		</int:header-enricher>
+		<int:delayer id="delayer" default-delay="10"/>
+	</int:chain>
+
+
 	<int:channel id="requestChannelFoo"/>
 	<int:channel id="requestChannelBar"/>
 	<int:channel id="requestChannelBaz"/>

spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests.java

@@ -19,6 +19,8 @@ package org.springframework.integration.gateway;
 import static org.hamcrest.Matchers.equalTo;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.Mockito.mock;
@@ -28,6 +30,7 @@ import static org.mockito.Mockito.verify;
 import java.lang.reflect.Method;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import org.hamcrest.Matchers;
 import org.junit.Test;
 import org.mockito.Mockito;
 
@@ -40,6 +43,7 @@ import org.springframework.integration.channel.DirectChannel;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.MessagingException;
+import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.support.MessageBuilder;
 
 /**
@@ -258,13 +262,30 @@ public class GatewayInterfaceTests {
 		assertTrue(called.get());
 	}
 
+	@Test
+	public void testLateReply() throws Exception {
+		ApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass());
+		Bar baz = ac.getBean(Bar.class);
+		String reply = baz.lateReply("hello");
+		assertNull(reply);
+		PollableChannel errorChannel = ac.getBean("errorChannel", PollableChannel.class);
+		Message<?> receive = errorChannel.receive(5000);
+		assertNotNull(receive);
+		MessagingException messagingException = (MessagingException) receive.getPayload();
+		assertThat(messagingException.getMessage(), Matchers.startsWith("Reply message received but the receiving thread has exited due to a timeout"));
+	}
+
 
 
 	public interface Foo {
+
 		@Gateway(requestChannel="requestChannelFoo")
 		public void foo(String payload);
 
 		public void baz(String payload);
+
+		public String lateReply(String payload);
+
 	}
 
 	public static interface Bar extends Foo {

spring-integration-core/src/test/resources/META-INF/spring.integration.properties

@@ -1,5 +1,5 @@
-#channels.autoCreate=false
-#channels.maxUnicastSubscribers=1
-#channels.maxBroadcastSubscribers=1
-messagingTemplate.lateReply.logging.level=error
-taskScheduler.poolSize=20
+#spring.integraton.channels.autoCreate=false
+#spring.integraton.channels.maxUnicastSubscribers=1
+#spring.integraton.channels.maxBroadcastSubscribers=1
+spring.integraton.taskScheduler.poolSize=20
+spring.integraton.messagingTemplate.throwExceptionOnLateReply=true