diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Concurrency.java b/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Concurrency.java deleted file mode 100644 index 4178855098..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Concurrency.java +++ /dev/null @@ -1,47 +0,0 @@ -/* - * Copyright 2002-2007 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. - */ - -package org.springframework.integration.annotation; - -import java.lang.annotation.Documented; -import java.lang.annotation.ElementType; -import java.lang.annotation.Inherited; -import java.lang.annotation.Retention; -import java.lang.annotation.RetentionPolicy; -import java.lang.annotation.Target; - -import org.springframework.integration.endpoint.ConcurrencyPolicy; - -/** - * Defines the {@link ConcurrencyPolicy} settings for a {@link MessageEndpoint @MessageEndpoint}. - * - * @author Mark Fisher - */ -@Target(ElementType.TYPE) -@Retention(RetentionPolicy.RUNTIME) -@Inherited -@Documented -public @interface Concurrency { - - int coreSize() default ConcurrencyPolicy.DEFAULT_CORE_SIZE; - - int maxSize() default ConcurrencyPolicy.DEFAULT_MAX_SIZE; - - int queueCapacity() default ConcurrencyPolicy.DEFAULT_QUEUE_CAPACITY; - - int keepAliveSeconds() default ConcurrencyPolicy.DEFAULT_KEEP_ALIVE_SECONDS; - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/ConcurrencyInterceptorParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/ConcurrencyInterceptorParser.java deleted file mode 100644 index ac6de7ec5d..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/ConcurrencyInterceptorParser.java +++ /dev/null @@ -1,76 +0,0 @@ -/* - * Copyright 2002-2008 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. - */ - -package org.springframework.integration.config; - -import org.w3c.dom.Element; - -import org.springframework.beans.factory.support.BeanDefinitionBuilder; -import org.springframework.beans.factory.support.BeanDefinitionReaderUtils; -import org.springframework.beans.factory.xml.ParserContext; -import org.springframework.integration.endpoint.ConcurrencyPolicy; -import org.springframework.integration.endpoint.interceptor.ConcurrencyInterceptor; -import org.springframework.util.StringUtils; - -/** - * Parser for the concurrency-interceptor. element. - * - * @author Mark Fisher - */ -public class ConcurrencyInterceptorParser implements BeanDefinitionRegisteringParser { - - public String parse(Element element, ParserContext parserContext) { - BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(ConcurrencyInterceptor.class); - String taskExecutorRef = element.getAttribute("task-executor"); - if (StringUtils.hasText(taskExecutorRef)) { - if (element.getAttributes().getLength() != 1) { - parserContext.getReaderContext().error("No other attributes are permitted when " - + "specifying a 'task-executor' reference on the element.", - parserContext.extractSource(element)); - } - builder.addConstructorArgReference(taskExecutorRef); - } - else { - ConcurrencyPolicy policy = this.parseConcurrencyPolicy(element); - builder.addConstructorArgValue(policy); - builder.addConstructorArgValue("concurrency-interceptor-"); - } - return BeanDefinitionReaderUtils.registerWithGeneratedName( - builder.getBeanDefinition(), parserContext.getRegistry()); - } - - private ConcurrencyPolicy parseConcurrencyPolicy(Element element) { - ConcurrencyPolicy policy = new ConcurrencyPolicy(); - String coreSize = element.getAttribute("core"); - String maxSize = element.getAttribute("max"); - String queueCapacity = element.getAttribute("queue-capacity"); - String keepAlive = element.getAttribute("keep-alive"); - if (StringUtils.hasText(coreSize)) { - policy.setCoreSize(Integer.parseInt(coreSize)); - } - if (StringUtils.hasText(maxSize)) { - policy.setMaxSize(Integer.parseInt(maxSize)); - } - if (StringUtils.hasText(queueCapacity)) { - policy.setQueueCapacity(Integer.parseInt(queueCapacity)); - } - if (StringUtils.hasText(keepAlive)) { - policy.setKeepAliveSeconds(Integer.parseInt(keepAlive)); - } - return policy; - } - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/EndpointInterceptorParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/EndpointInterceptorParser.java index 89ee1c1e61..70dbf47efa 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/EndpointInterceptorParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/EndpointInterceptorParser.java @@ -26,10 +26,7 @@ public class EndpointInterceptorParser extends AbstractInterceptorParser { @Override protected Map getParserMap() { - Map parsers = new HashMap(); - parsers.put("transaction-interceptor", new TransactionInterceptorParser()); - parsers.put("concurrency-interceptor", new ConcurrencyInterceptorParser()); - return parsers; + return new HashMap(); } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/TransactionInterceptorParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/TransactionInterceptorParser.java deleted file mode 100644 index dbb41e7e24..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/TransactionInterceptorParser.java +++ /dev/null @@ -1,67 +0,0 @@ -/* - * Copyright 2002-2008 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. - */ - -package org.springframework.integration.config; - -import org.w3c.dom.Element; - -import org.springframework.beans.factory.support.BeanDefinitionBuilder; -import org.springframework.beans.factory.support.BeanDefinitionReaderUtils; -import org.springframework.beans.factory.xml.ParserContext; -import org.springframework.integration.endpoint.interceptor.TransactionInterceptor; -import org.springframework.transaction.interceptor.RuleBasedTransactionAttribute; -import org.springframework.util.StringUtils; - -/** - * Parser for the transaction-interceptor element. - * - * @author Mark Fisher - */ -public class TransactionInterceptorParser implements BeanDefinitionRegisteringParser { - - @SuppressWarnings("unchecked") - public String parse(Element element, ParserContext parserContext) { - BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(TransactionInterceptor.class); - String txManagerRef = element.getAttribute("transaction-manager"); - if (!StringUtils.hasText(txManagerRef)) { - txManagerRef = "transactionManager"; - } - builder.addConstructorArgReference(txManagerRef); - String propagation = element.getAttribute("propagation"); - String isolation = element.getAttribute("isolation"); - String timeout = element.getAttribute("timeout"); - String readOnly = element.getAttribute("read-only"); - if (StringUtils.hasText(propagation)) { - builder.addPropertyValue("propagationBehaviorName", RuleBasedTransactionAttribute.PREFIX_PROPAGATION + propagation); - } - if (StringUtils.hasText(isolation)) { - builder.addPropertyValue("isolationLevelName", RuleBasedTransactionAttribute.PREFIX_ISOLATION + isolation); - } - if (StringUtils.hasText(timeout)) { - try { - builder.addPropertyValue("timeout", Integer.parseInt(timeout)); - } - catch (NumberFormatException ex) { - parserContext.getReaderContext().error("Timeout must be an integer value: [" + timeout + "]", element); - } - } - if (StringUtils.hasText(readOnly)) { - builder.addPropertyValue("readOnly", Boolean.valueOf(readOnly)); - } - return BeanDefinitionReaderUtils.registerWithGeneratedName(builder.getBeanDefinition(), parserContext.getRegistry()); - } - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/HandlerAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/HandlerAnnotationPostProcessor.java index a6330c84d6..f6a43b51eb 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/HandlerAnnotationPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/HandlerAnnotationPostProcessor.java @@ -31,17 +31,14 @@ import org.springframework.core.annotation.Order; import org.springframework.integration.ConfigurationException; import org.springframework.integration.aggregator.AggregatorMessageHandlerCreator; import org.springframework.integration.annotation.Aggregator; -import org.springframework.integration.annotation.Concurrency; import org.springframework.integration.annotation.Handler; import org.springframework.integration.annotation.Router; import org.springframework.integration.annotation.Splitter; import org.springframework.integration.annotation.Transformer; import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.ChannelRegistryAware; -import org.springframework.integration.endpoint.ConcurrencyPolicy; import org.springframework.integration.endpoint.MessageEndpoint; import org.springframework.integration.endpoint.DefaultEndpoint; -import org.springframework.integration.endpoint.interceptor.ConcurrencyInterceptor; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.handler.MessageHandlerChain; import org.springframework.integration.handler.config.DefaultMessageHandlerCreator; @@ -144,14 +141,6 @@ public class HandlerAnnotationPostProcessor extends AbstractAnnotationMethodPost if (schedule != null) { endpoint.setSchedule(schedule); } - Concurrency concurrencyAnnotation = AnnotationUtils.findAnnotation(originalBeanClass, Concurrency.class); - if (concurrencyAnnotation != null) { - ConcurrencyPolicy concurrencyPolicy = new ConcurrencyPolicy( - concurrencyAnnotation.coreSize(), concurrencyAnnotation.maxSize()); - concurrencyPolicy.setKeepAliveSeconds(concurrencyAnnotation.keepAliveSeconds()); - concurrencyPolicy.setQueueCapacity(concurrencyAnnotation.queueCapacity()); - endpoint.addInterceptor(new ConcurrencyInterceptor(concurrencyPolicy, beanName)); - } return endpoint; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/DefaultEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/DefaultEndpoint.java index 8acd2f89cb..15cef2ecd0 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/DefaultEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/DefaultEndpoint.java @@ -110,7 +110,26 @@ public class DefaultEndpoint extends AbstractRequestRe @Override public final Message handleRequestMessage(Message requestMessage) { - return this.doHandleRequestMessage(requestMessage, 0); + if (requestMessage == null || requestMessage.getPayload() == null) { + return null; + } + for (EndpointInterceptor interceptor : this.interceptors) { + requestMessage = interceptor.preHandle(requestMessage); + if (requestMessage == null) { + return null; + } + } + if (!this.supports(requestMessage)) { + throw new MessageRejectedException(requestMessage, "unsupported message"); + } + Message replyMessage = this.handler.handle(requestMessage); + for (int i = this.interceptors.size() - 1; i >= 0; i--) { + EndpointInterceptor interceptor = this.interceptors.get(i); + if (interceptor != null) { + replyMessage = interceptor.postHandle(replyMessage); + } + } + return replyMessage; } /** @@ -155,38 +174,6 @@ public class DefaultEndpoint extends AbstractRequestRe } } - private Message doHandleRequestMessage(Message requestMessage, final int index) { - if (requestMessage == null || requestMessage.getPayload() == null) { - return null; - } - if (index == 0) { - for (EndpointInterceptor interceptor : this.interceptors) { - requestMessage = interceptor.preHandle(requestMessage); - if (requestMessage == null) { - return null; - } - } - } - if (index == this.interceptors.size()) { - if (!this.supports(requestMessage)) { - throw new MessageRejectedException(requestMessage, "unsupported message"); - } - Message replyMessage = this.handler.handle(requestMessage); - for (int i = index - 1; i >= 0; i--) { - EndpointInterceptor interceptor = this.interceptors.get(i); - replyMessage = interceptor.postHandle(replyMessage); - } - return replyMessage; - } - EndpointInterceptor nextInterceptor = this.interceptors.get(index); - return nextInterceptor.aroundHandle(requestMessage, new MessageHandler() { - @SuppressWarnings("unchecked") - public Message handle(Message message) { - return doHandleRequestMessage(message, index + 1); - } - }); - } - protected boolean supports(Message message) { if (this.selector != null && !this.selector.accept(message)) { if (logger.isDebugEnabled()) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointInterceptor.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointInterceptor.java index ac949d7ad8..b015b8f24b 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointInterceptor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointInterceptor.java @@ -16,7 +16,6 @@ package org.springframework.integration.endpoint; -import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; /** @@ -26,8 +25,6 @@ public interface EndpointInterceptor { Message preHandle(Message requestMessage); - Message aroundHandle(Message requestMessage, MessageHandler handler); - Message postHandle(Message replyMessage); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/ConcurrencyInterceptor.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/ConcurrencyInterceptor.java deleted file mode 100644 index f61236083c..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/ConcurrencyInterceptor.java +++ /dev/null @@ -1,157 +0,0 @@ -/* - * Copyright 2002-2008 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. - */ - -package org.springframework.integration.endpoint.interceptor; - -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.Callable; -import java.util.concurrent.Executor; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.FutureTask; -import java.util.concurrent.LinkedBlockingQueue; -import java.util.concurrent.SynchronousQueue; -import java.util.concurrent.ThreadPoolExecutor; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - -import org.springframework.beans.factory.DisposableBean; -import org.springframework.beans.factory.InitializingBean; -import org.springframework.core.task.TaskExecutor; -import org.springframework.integration.channel.ChannelRegistry; -import org.springframework.integration.channel.ChannelRegistryAware; -import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.MessagePublishingErrorHandler; -import org.springframework.integration.endpoint.ConcurrencyPolicy; -import org.springframework.integration.endpoint.EndpointInterceptor; -import org.springframework.integration.handler.MessageHandler; -import org.springframework.integration.handler.MessageHandlerRejectedExecutionException; -import org.springframework.integration.message.AsyncMessage; -import org.springframework.integration.message.Message; -import org.springframework.integration.util.ErrorHandler; -import org.springframework.scheduling.concurrent.ConcurrentTaskExecutor; -import org.springframework.scheduling.concurrent.CustomizableThreadFactory; -import org.springframework.util.Assert; - -/** - * An {@link EndpointInterceptor} implementation that delegates to a - * {@link TaskExecutor} for concurrent, asynchronous message handling. - * - * @author Mark Fisher - */ -public class ConcurrencyInterceptor extends EndpointInterceptorAdapter - implements DisposableBean, InitializingBean, ChannelRegistryAware { - - private final Log logger = LogFactory.getLog(this.getClass()); - - private final TaskExecutor executor; - - private volatile ErrorHandler errorHandler; - - private volatile ChannelRegistry channelRegistry; - - - public ConcurrencyInterceptor(TaskExecutor executor) { - Assert.notNull(executor, "TaskExecutor must not be null"); - this.executor = executor; - } - - public ConcurrencyInterceptor(ConcurrencyPolicy concurrencyPolicy, String threadPrefix) { - Assert.notNull(concurrencyPolicy, "ConcurrencyPolicy must not be null"); - int core = concurrencyPolicy.getCoreSize(); - int max = concurrencyPolicy.getMaxSize(); - int keepAlive = concurrencyPolicy.getKeepAliveSeconds(); - int capacity = concurrencyPolicy.getQueueCapacity(); - BlockingQueue queue = (capacity > 0) ? - new LinkedBlockingQueue(capacity) : new SynchronousQueue(); - ThreadPoolExecutor tpe = new ThreadPoolExecutor(core, max, keepAlive, TimeUnit.SECONDS, queue); - tpe.setThreadFactory(new CustomizableThreadFactory(threadPrefix)); - tpe.setRejectedExecutionHandler(new CallerRunsPolicy()); - this.executor = new ConcurrentTaskExecutor(tpe); - } - - - public void setErrorHandler(ErrorHandler errorHandler) { - this.errorHandler = errorHandler; - } - - public void setChannelRegistry(ChannelRegistry channelRegistry) { - this.channelRegistry = channelRegistry; - } - - public void afterPropertiesSet() throws Exception { - if (this.errorHandler == null && this.channelRegistry != null) { - MessageChannel errorChannel = this.channelRegistry.lookupChannel( - ChannelRegistry.ERROR_CHANNEL_NAME); - if (errorChannel != null) { - this.errorHandler = new MessagePublishingErrorHandler(errorChannel); - } - } - } - - public void destroy() throws Exception { - if (this.executor instanceof DisposableBean) { - ((DisposableBean) this.executor).destroy(); - } - if (this.executor instanceof ConcurrentTaskExecutor) { - Executor innerExecutor = ((ConcurrentTaskExecutor) this.executor).getConcurrentExecutor(); - if (innerExecutor instanceof ExecutorService) { - ExecutorService executorService = (ExecutorService) innerExecutor; - executorService.shutdown(); - if (!executorService.awaitTermination(10, TimeUnit.SECONDS)) { - executorService.shutdownNow(); - } - } - } - } - - @Override - @SuppressWarnings("unchecked") - public Message aroundHandle(final Message requestMessage, final MessageHandler handler) { - try { - FutureTask> task = new FutureTask>(new Callable>() { - public Message call() throws Exception { - try { - return handler.handle(requestMessage); - } - catch (Exception e) { - if (logger.isDebugEnabled()) { - logger.debug("error occurred in handler execution", e); - } - if (errorHandler != null) { - errorHandler.handle(e); - } - else { - if (logger.isWarnEnabled() && !logger.isDebugEnabled()) { - logger.warn("error occurred in handler execution", e); - } - throw e; - } - } - return null; - } - }); - this.executor.execute(task); - return new AsyncMessage(task); - } - catch (RuntimeException e) { - throw new MessageHandlerRejectedExecutionException(requestMessage, e); - } - } - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/EndpointInterceptorAdapter.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/EndpointInterceptorAdapter.java index d34a5bc939..3e35974982 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/EndpointInterceptorAdapter.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/EndpointInterceptorAdapter.java @@ -17,7 +17,6 @@ package org.springframework.integration.endpoint.interceptor; import org.springframework.integration.endpoint.EndpointInterceptor; -import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; /** @@ -31,10 +30,6 @@ public class EndpointInterceptorAdapter implements EndpointInterceptor { return requestMessage; } - public Message aroundHandle(Message requestMessage, MessageHandler handler) { - return handler.handle(requestMessage); - } - public Message postHandle(Message replyMessage) { return replyMessage; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/TransactionInterceptor.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/TransactionInterceptor.java deleted file mode 100644 index 3108e52052..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/TransactionInterceptor.java +++ /dev/null @@ -1,100 +0,0 @@ -/* - * Copyright 2002-2008 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. - */ - -package org.springframework.integration.endpoint.interceptor; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - -import org.springframework.beans.factory.InitializingBean; -import org.springframework.integration.ConfigurationException; -import org.springframework.integration.endpoint.EndpointInterceptor; -import org.springframework.integration.handler.MessageHandler; -import org.springframework.integration.message.Message; -import org.springframework.transaction.PlatformTransactionManager; -import org.springframework.transaction.TransactionStatus; -import org.springframework.transaction.support.TransactionCallback; -import org.springframework.transaction.support.TransactionTemplate; - -/** - * An {@link EndpointInterceptor} implementation that provides transactional - * behavior with a {@link PlatformTransactionManager}. - * - * @author Mark Fisher - */ -public class TransactionInterceptor extends EndpointInterceptorAdapter implements InitializingBean { - - private final Log logger = LogFactory.getLog(this.getClass()); - - private final PlatformTransactionManager transactionManager; - - private volatile TransactionTemplate transactionTemplate; - - private volatile String propagationBehaviorName = "PROPAGATION_REQUIRED"; - - private volatile String isolationLevelName = "ISOLATION_DEFAULT"; - - private volatile int timeout = -1; - - private volatile boolean readOnly = false; - - - public TransactionInterceptor(PlatformTransactionManager transactionManager) { - this.transactionManager = transactionManager; - } - - - public void setPropagationBehaviorName(String propagationBehaviorName) { - this.propagationBehaviorName = propagationBehaviorName; - } - - public void setIsolationLevelName(String isolationLevelName) { - this.isolationLevelName = isolationLevelName; - } - - public void setTimeout(int timeout) { - this.timeout = timeout; - } - - public void setReadOnly(boolean readOnly) { - this.readOnly = readOnly; - } - - public void afterPropertiesSet() { - TransactionTemplate template = new TransactionTemplate(this.transactionManager); - template.setPropagationBehaviorName(this.propagationBehaviorName); - template.setIsolationLevelName(this.isolationLevelName); - template.setTimeout(this.timeout); - template.setReadOnly(this.readOnly); - this.transactionTemplate = template; - } - - @Override - public Message aroundHandle(final Message message, final MessageHandler handler) { - if (this.transactionTemplate == null) { - throw new ConfigurationException("TransactionInterceptor has not been initialized"); - } - return (Message) this.transactionTemplate.execute(new TransactionCallback() { - public Object doInTransaction(TransactionStatus status) { - if (logger.isDebugEnabled()) { - logger.debug("Invoking handler '" + handler + "' within transaction [" + status + "]"); - } - return handler.handle(message); - } - }); - } - -} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java index 3f3ff8e848..14af210c32 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java @@ -25,22 +25,18 @@ import java.lang.annotation.ElementType; import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; -import java.util.List; import java.util.concurrent.CountDownLatch; -import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; import org.junit.Test; import org.springframework.aop.framework.ProxyFactory; -import org.springframework.beans.DirectFieldAccessor; import org.springframework.context.support.AbstractApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.core.annotation.Order; import org.springframework.integration.ConfigurationException; import org.springframework.integration.annotation.ChannelAdapter; -import org.springframework.integration.annotation.Concurrency; import org.springframework.integration.annotation.Handler; import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.annotation.MessageTarget; @@ -56,14 +52,12 @@ import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.endpoint.EndpointInterceptor; import org.springframework.integration.endpoint.DefaultEndpoint; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; import org.springframework.integration.message.StringMessage; import org.springframework.integration.scheduling.PollingSchedule; import org.springframework.integration.scheduling.Schedule; -import org.springframework.scheduling.concurrent.ConcurrentTaskExecutor; /** * @author Mark Fisher @@ -165,28 +159,6 @@ public class MessagingAnnotationPostProcessorTests { messageBus.stop(); } - @Test - @SuppressWarnings("unchecked") - public void testConcurrencyAnnotationWithValues() { - MessageBus messageBus = new DefaultMessageBus(); - MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); - postProcessor.afterPropertiesSet(); - ConcurrencyAnnotationTestBean testBean = new ConcurrencyAnnotationTestBean(); - postProcessor.postProcessAfterInitialization(testBean, "testBean"); - DefaultEndpoint endpoint = (DefaultEndpoint) messageBus.lookupEndpoint("testBean.MessageHandler.endpoint"); - DirectFieldAccessor accessor = new DirectFieldAccessor(endpoint); - List interceptors = (List) accessor.getPropertyValue("interceptors"); - assertEquals(1, interceptors.size()); - EndpointInterceptor interceptor = interceptors.get(0); - accessor = new DirectFieldAccessor(interceptor); - ConcurrentTaskExecutor cte = (ConcurrentTaskExecutor) accessor.getPropertyValue("executor"); - ThreadPoolExecutor executor = (ThreadPoolExecutor) cte.getConcurrentExecutor(); - assertEquals(17, executor.getCorePoolSize()); - assertEquals(42, executor.getMaximumPoolSize()); - assertEquals(123, executor.getKeepAliveTime(TimeUnit.SECONDS)); - assertEquals(11, executor.getQueue().remainingCapacity()); - } - @Test(expected=IllegalArgumentException.class) public void testPostProcessorWithNullMessageBus() { new MessagingAnnotationPostProcessor(null); @@ -450,17 +422,6 @@ public class MessagingAnnotationPostProcessorTests { } - @MessageEndpoint(input="inputChannel") - @Concurrency(coreSize=17, maxSize=42, keepAliveSeconds=123, queueCapacity=11) - private static class ConcurrencyAnnotationTestBean { - - @Handler - public Message handle(Message message) { - return null; - } - } - - @MessageEndpoint(input="inputChannel") private static class ChannelRegistryAwareTestBean implements ChannelRegistryAware {