diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aop/MessagePublishingInterceptor.java b/spring-integration-core/src/main/java/org/springframework/integration/aop/MessagePublishingInterceptor.java index 41bed17ec8..8815d2a2f9 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aop/MessagePublishingInterceptor.java +++ b/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 { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/DefaultConfiguringBeanFactoryPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/DefaultConfiguringBeanFactoryPostProcessor.java index 7a600c0436..df4b0b6a76 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/DefaultConfiguringBeanFactoryPostProcessor.java +++ b/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()); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationProperties.java b/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationProperties.java index a02eb0cfe0..7c587ce930 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationProperties.java +++ b/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()); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java b/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java index 9839dcd258..887c593009 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java +++ b/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); } /** diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java index 0264edad90..09254dd60d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java +++ b/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()); + } } /** diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java index a3f7210190..5ab90dea61 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java +++ b/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 diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java index efcc995337..f25b2fde35 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java +++ b/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; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/ErrorMessageSendingRecoverer.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/ErrorMessageSendingRecoverer.java index fcbdca902c..aff49d10cb 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/ErrorMessageSendingRecoverer.java +++ b/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 { +public class ErrorMessageSendingRecoverer implements RecoveryCallback, 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 { 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) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/ExpressionEvaluatingRequestHandlerAdvice.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/ExpressionEvaluatingRequestHandlerAdvice.java index 2e946ada24..45672ef5ff 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/ExpressionEvaluatingRequestHandlerAdvice.java +++ b/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 { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java index facfd93dc0..8a3b233b1e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java +++ b/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. diff --git a/spring-integration-core/src/main/resources/META-INF/spring.integration.default.properties b/spring-integration-core/src/main/resources/META-INF/spring.integration.default.properties index 3633615b97..e414251dae 100644 --- a/spring-integration-core/src/main/resources/META-INF/spring.integration.default.properties +++ b/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 diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersOverrideDefaultTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersOverrideDefaultTests-context.xml index b4a39d9398..1fcfdfc3a4 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersOverrideDefaultTests-context.xml +++ b/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"> - 456 - 789 + 456 + 789 diff --git a/spring-integration-core/src/test/java/org/springframework/integration/context/IntegrationContextTests.java b/spring-integration-core/src/test/java/org/springframework/integration/context/IntegrationContextTests.java index 3ad521a9bc..61965697b7 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/context/IntegrationContextTests.java +++ b/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")); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests-context.xml index ce665b5115..e39484b141 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests-context.xml +++ b/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"> + + + + + default-request-channel="requestChannelBaz" + error-channel="errorChannel"> + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests.java index 8078a87536..0f721460f4 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests.java +++ b/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 { diff --git a/spring-integration-core/src/test/resources/META-INF/spring.integration.properties b/spring-integration-core/src/test/resources/META-INF/spring.integration.properties index bd78272249..cf4ecc3cb8 100644 --- a/spring-integration-core/src/test/resources/META-INF/spring.integration.properties +++ b/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