Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Removed the aroundHandle() method from EndpointInterceptor now that the endpoints are simplified ("normal" AOP interception can easily be applied to the invoked target object). Removed the @Concurrency annotation since it was being applied by the endpointInterceptor instead of the poller and since the thread-pool executor is only one of the available taskExecutor options (may provide a 'taskExecutor' reference for @Poller instead). Removed the ConcurrencyInterceptor and TransactionInterceptor implementations.

Browse Source
This commit is contained in:
Mark Fisher
2008-08-30 17:42:45 +00:00
parent 41b3a764ab
commit bdfc98f966
11 changed files with 21 additions and 542 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
org.springframework.integration/src/main/java/org/springframework/integration/annotation/Concurrency.java
View File

@@ -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;
}

76
org.springframework.integration/src/main/java/org/springframework/integration/config/ConcurrencyInterceptorParser.java
View File

@@ -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 <em>concurrency-interceptor</em>. 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 <concurrency-interceptor/> 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;
}
}

5
org.springframework.integration/src/main/java/org/springframework/integration/config/EndpointInterceptorParser.java
View File

@@ -26,10 +26,7 @@ public class EndpointInterceptorParser extends AbstractInterceptorParser {
@Override
protected Map<String, BeanDefinitionRegisteringParser> getParserMap() {
Map<String, BeanDefinitionRegisteringParser> parsers = new HashMap<String, BeanDefinitionRegisteringParser>();
parsers.put("transaction-interceptor", new TransactionInterceptorParser());
parsers.put("concurrency-interceptor", new ConcurrencyInterceptorParser());
return parsers;
return new HashMap<String, BeanDefinitionRegisteringParser>();
}
}

67
org.springframework.integration/src/main/java/org/springframework/integration/config/TransactionInterceptorParser.java
View File

@@ -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 <em>transaction-interceptor</em> 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());
}
}

11
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/HandlerAnnotationPostProcessor.java
View File

@@ -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;
}

53
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/DefaultEndpoint.java
View File

@@ -110,7 +110,26 @@ public class DefaultEndpoint<T extends MessageHandler> 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<T extends MessageHandler> 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()) {

3
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointInterceptor.java
View File

@@ -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);
}

157
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/ConcurrencyInterceptor.java
View File

@@ -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<Runnable> queue = (capacity > 0) ?
new LinkedBlockingQueue<Runnable>(capacity) : new SynchronousQueue<Runnable>();
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<Message<?>> task = new FutureTask<Message<?>>(new Callable<Message<?>>() {
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);
}
}
}

5
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/EndpointInterceptorAdapter.java
View File

@@ -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;
}

100
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/TransactionInterceptor.java
View File

@@ -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);
}
});
}
}

39
org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java
View File

@@ -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<EndpointInterceptor> interceptors = (List<EndpointInterceptor>) 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 {