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

channel implementations now extend IntegrationObjectSupport, and MethodInvokingMessageProcessor now provides a setter for a ConversionService

Browse Source
This commit is contained in:
Mark Fisher
2010-03-05 20:52:30 +00:00
parent ed5359eee1
commit 590fd3bc5a
11 changed files with 141 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files

66
org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java
View File

@@ -22,10 +22,8 @@ import java.util.concurrent.CopyOnWriteArrayList;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.core.convert.ConversionService;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessagingException;
@@ -43,41 +41,20 @@ import org.springframework.util.StringUtils;
*
* @author Mark Fisher
*/
public abstract class AbstractMessageChannel implements MessageChannel, BeanFactoryAware, BeanNameAware {
public abstract class AbstractMessageChannel extends IntegrationObjectSupport implements MessageChannel {
private final Log logger = LogFactory.getLog(this.getClass());
private volatile String name;
private volatile Class<?>[] datatypes = new Class<?>[] { Object.class };
private volatile ConversionService conversionService;
private volatile BeanFactory beanFactory;
private final ComponentMetadata metadata = new ComponentMetadata();
private final ChannelInterceptorList interceptors = new ChannelInterceptorList();
public AbstractMessageChannel() {
this.metadata.setComponentType("channel");
}
/**
* Set the name of this channel. This will be invoked automatically whenever
* the channel is configured explicitly with a bean definition.
*/
public void setBeanName(String name) {
this.name = name;
this.metadata.setComponentName(name);
}
/**
* Return the name of this channel.
*/
public String getName() {
return this.name;
return this.getBeanName();
}
/**
@@ -114,29 +91,13 @@ public abstract class AbstractMessageChannel implements MessageChannel, BeanFact
* Specify the {@link ConversionService} to use when trying to convert to
* one of this channel's supported datatypes for a Message whose payload
* does not already match. If this property is not set explicitly but
* the channel is managed within a context, it will fallback to a bean
* named "conversionService" defined within that context.
* the channel is managed within a context, it will attempt to locate a
* bean named "integrationConversionService" defined within that context.
* Finally, if that bean is not available, it will fallback to the
* "conversionService" bean, if available.
*/
public void setConversionService(ConversionService conversionService) {
this.conversionService = conversionService;
}
public void setBeanFactory(BeanFactory beanFactory) {
this.beanFactory = beanFactory;
}
private ConversionService getConversionService() {
if (this.conversionService == null && this.beanFactory != null) {
if (this.beanFactory.containsBean("conversionService")) {
this.conversionService = this.beanFactory.getBean("conversionService", ConversionService.class);
}
else if (logger.isWarnEnabled()) {
logger.warn("Unable to attempt conversion of Message payload types. Datatype channel '" +
this.getName() + "' has no explicit ConversionService reference, " +
"and there is no 'conversionService' bean within the context.");
}
}
return this.conversionService;
super.setConversionService(conversionService);
}
/**
@@ -146,6 +107,11 @@ public abstract class AbstractMessageChannel implements MessageChannel, BeanFact
return this.interceptors;
}
@Override
protected void populateComponentMetadata(ComponentMetadata metadata) {
metadata.setComponentType("channel");
}
/**
* Send a message on this channel. If the channel is at capacity, this
* method will block until either space becomes available or the sending
@@ -178,7 +144,7 @@ public abstract class AbstractMessageChannel implements MessageChannel, BeanFact
Assert.notNull(message, "message must not be null");
Assert.notNull(message.getPayload(), "message payload must not be null");
message = this.convertPayloadIfNecessary(message);
message.getHeaders().getHistory().addEvent(this.metadata);
message.getHeaders().getHistory().addEvent(this.getComponentMetadata());
message = this.interceptors.preSend(message, this);
if (message == null) {
return false;
@@ -220,10 +186,6 @@ public abstract class AbstractMessageChannel implements MessageChannel, BeanFact
"], but received [" + message.getPayload().getClass() + "]");
}
public String toString() {
return (this.name != null) ? this.name : super.toString();
}
/**
* Subclasses must implement this method. A non-negative timeout indicates
* how long to wait if the channel is at capacity (if the value is 0, it

10
org.springframework.integration/src/main/java/org/springframework/integration/channel/ExecutorChannel.java
View File

@@ -18,8 +18,6 @@ package org.springframework.integration.channel;
import java.util.concurrent.Executor;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.dispatcher.LoadBalancingStrategy;
import org.springframework.integration.dispatcher.UnicastingDispatcher;
@@ -42,7 +40,7 @@ import org.springframework.util.ErrorHandler;
* @author Mark Fisher
* @since 1.0.3
*/
public class ExecutorChannel extends AbstractSubscribableChannel implements BeanFactoryAware {
public class ExecutorChannel extends AbstractSubscribableChannel {
private volatile UnicastingDispatcher dispatcher;
@@ -94,9 +92,11 @@ public class ExecutorChannel extends AbstractSubscribableChannel implements Bean
return this.dispatcher;
}
public void setBeanFactory(BeanFactory beanFactory) {
@Override
public final void onInit() {
if (!(this.executor instanceof ErrorHandlingTaskExecutor)) {
ErrorHandler errorHandler = new MessagePublishingErrorHandler(new BeanFactoryChannelResolver(beanFactory));
ErrorHandler errorHandler = new MessagePublishingErrorHandler(
new BeanFactoryChannelResolver(this.getBeanFactory()));
this.executor = new ErrorHandlingTaskExecutor(this.executor, errorHandler);
}
this.dispatcher = new UnicastingDispatcher(this.executor);

12
org.springframework.integration/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java
View File

@@ -18,8 +18,6 @@ package org.springframework.integration.channel;
import java.util.concurrent.Executor;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.integration.dispatcher.BroadcastingDispatcher;
import org.springframework.integration.util.ErrorHandlingTaskExecutor;
import org.springframework.util.ErrorHandler;
@@ -29,7 +27,7 @@ import org.springframework.util.ErrorHandler;
*
* @author Mark Fisher
*/
public class PublishSubscribeChannel extends AbstractSubscribableChannel implements BeanFactoryAware {
public class PublishSubscribeChannel extends AbstractSubscribableChannel {
private volatile BroadcastingDispatcher dispatcher;
@@ -102,13 +100,15 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel impleme
}
/**
* Callback method for the {@link BeanFactoryAware} interface.
* Callback method for initialization.
*/
public void setBeanFactory(BeanFactory beanFactory) {
@Override
public final void onInit() {
if (this.executor != null) {
if (!(this.executor instanceof ErrorHandlingTaskExecutor)) {
if (this.errorHandler == null) {
this.errorHandler = new MessagePublishingErrorHandler(new BeanFactoryChannelResolver(beanFactory));
this.errorHandler = new MessagePublishingErrorHandler(
new BeanFactoryChannelResolver(this.getBeanFactory()));
}
this.executor = new ErrorHandlingTaskExecutor(this.executor, this.errorHandler);
}

72
org.springframework.integration/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2009 the original author or authors.
* Copyright 2002-2010 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.
@@ -21,9 +21,14 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.BeanInitializationException;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.convert.ConversionService;
import org.springframework.integration.channel.BeanFactoryChannelResolver;
import org.springframework.integration.channel.ChannelResolver;
import org.springframework.integration.support.ComponentMetadata;
import org.springframework.integration.support.ComponentMetadataProvider;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.util.Assert;
@@ -38,7 +43,7 @@ import org.springframework.util.Assert;
*
* @author Mark Fisher
*/
public abstract class IntegrationObjectSupport implements BeanNameAware, BeanFactoryAware {
public abstract class IntegrationObjectSupport implements ComponentMetadataProvider, BeanNameAware, BeanFactoryAware, InitializingBean {
/** Logger that is available to subclasses */
protected final Log logger = LogFactory.getLog(getClass());
@@ -51,8 +56,12 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, BeanFac
private volatile TaskScheduler taskScheduler;
private volatile ConversionService conversionService;
public void setBeanName(String beanName) {
private final ComponentMetadata componentMetadata = new ComponentMetadata();
public final void setBeanName(String beanName) {
this.beanName = beanName;
}
@@ -60,25 +69,50 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, BeanFac
return this.beanName;
}
public ComponentMetadata getComponentMetadata() {
return this.componentMetadata;
}
public final void setBeanFactory(BeanFactory beanFactory) {
Assert.notNull(beanFactory, "beanFactory must not be null");
this.beanFactory = beanFactory;
this.channelResolver = new BeanFactoryChannelResolver(beanFactory);
TaskScheduler taskScheduler = IntegrationContextUtils.getTaskScheduler(beanFactory);
if (taskScheduler != null) {
this.taskScheduler = taskScheduler;
}
public final void afterPropertiesSet() {
this.componentMetadata.setComponentName(this.beanName);
this.populateComponentMetadata(this.componentMetadata);
try {
this.onInit();
}
catch (Exception e) {
if (e instanceof RuntimeException) {
throw (RuntimeException) e;
}
throw new BeanInitializationException("failed to initialize", e);
}
}
protected BeanFactory getBeanFactory() {
/**
* Subclasses may implement this for initialization logic.
*/
protected void onInit() throws Exception {
}
protected final BeanFactory getBeanFactory() {
return this.beanFactory;
}
protected ChannelResolver getChannelResolver() {
if (this.channelResolver == null && this.beanFactory != null) {
this.channelResolver = new BeanFactoryChannelResolver(this.beanFactory);
}
return this.channelResolver;
}
protected TaskScheduler getTaskScheduler() {
if (this.taskScheduler == null && this.beanFactory != null) {
this.taskScheduler = IntegrationContextUtils.getTaskScheduler(this.beanFactory);
}
return this.taskScheduler;
}
@@ -87,6 +121,28 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, BeanFac
this.taskScheduler = taskScheduler;
}
protected final ConversionService getConversionService() {
if (this.conversionService == null && this.beanFactory != null) {
this.conversionService = IntegrationContextUtils.getConversionService(this.beanFactory);
if (this.conversionService == null && logger.isWarnEnabled()) {
logger.warn("Unable to attempt conversion of Message payload types. Component '" +
this.getBeanName() + "' has no explicit ConversionService reference, " +
"and there is no 'integrationConversionService' or 'conversionService' " +
"bean within the context.");
}
}
return this.conversionService;
}
protected void setConversionService(ConversionService conversionService) {
this.conversionService = conversionService;
}
/**
* Subclasses may override this to add attributes to the {@link ComponentMetadata}.
*/
protected void populateComponentMetadata(ComponentMetadata metadata) {
}
@Override
public String toString() {

18
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2009 the original author or authors.
* Copyright 2002-2010 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,8 +18,6 @@ package org.springframework.integration.endpoint;
import java.util.concurrent.locks.ReentrantLock;
import org.springframework.beans.factory.BeanInitializationException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.SmartLifecycle;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.scheduling.TaskScheduler;
@@ -36,7 +34,7 @@ import org.springframework.scheduling.TaskScheduler;
*
* @author Mark Fisher
*/
public abstract class AbstractEndpoint extends IntegrationObjectSupport implements SmartLifecycle, InitializingBean {
public abstract class AbstractEndpoint extends IntegrationObjectSupport implements SmartLifecycle {
private volatile boolean autoStartup = true;
@@ -59,15 +57,6 @@ public abstract class AbstractEndpoint extends IntegrationObjectSupport implemen
super.setTaskScheduler(taskScheduler);
}
public final void afterPropertiesSet() {
try {
this.onInit();
}
catch (Exception e) {
throw new BeanInitializationException("failed to initialize", e);
}
}
// SmartLifecycle implementation
public final boolean isAutoStartup() {
@@ -131,9 +120,6 @@ public abstract class AbstractEndpoint extends IntegrationObjectSupport implemen
}
}
protected void onInit() throws Exception {
}
/**
* Subclasses must implement this method with the start behavior.
* This method will be invoked while holding the {@link #lifecycleLock}.

9
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java
View File

@@ -48,8 +48,6 @@ public abstract class MessageProducerSupport extends AbstractEndpoint {
@Override
protected void onInit() {
Assert.notNull(this.outputChannel, "outputChannel is required");
this.componentMetadata.setComponentName(this.getBeanName());
this.populateComponentMetadata(this.componentMetadata);
}
protected boolean sendMessage(Message<?> message) {
@@ -59,11 +57,4 @@ public abstract class MessageProducerSupport extends AbstractEndpoint {
return this.channelTemplate.send(message, this.outputChannel);
}
/**
* Subclasses may override this no-op method to add attributes to this
* adapter's {@link ComponentMetadata}.
*/
protected void populateComponentMetadata(ComponentMetadata metadata) {
}
}

4
org.springframework.integration/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java
View File

@@ -143,7 +143,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Factory
}
@Override
protected void onInit() throws Exception {
protected void onInit() {
synchronized (this.initializationMonitor) {
if (this.initialized) {
return;
@@ -232,7 +232,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Factory
throw originalException;
}
private MessagingGateway createGatewayForMethod(Method method) throws Exception {
private MessagingGateway createGatewayForMethod(Method method) {
SimpleMessagingGateway gateway = new SimpleMessagingGateway(
new ArgumentArrayMessageMapper(method, this.getBeanName()), new SimpleMessageMapper());
if (this.getTaskScheduler() != null) {

35
org.springframework.integration/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java
View File

@@ -20,9 +20,13 @@ import java.util.List;
import org.springframework.beans.BeanWrapperImpl;
import org.springframework.beans.PropertyAccessor;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.core.Ordered;
import org.springframework.integration.channel.BeanFactoryChannelResolver;
import org.springframework.integration.channel.ChannelResolver;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.endpoint.EventDrivenConsumer;
@@ -63,7 +67,7 @@ import org.springframework.util.Assert;
* @author Mark Fisher
* @author Iwein Fuld
*/
public class MessageHandlerChain extends IntegrationObjectSupport implements MessageHandler, Ordered {
public class MessageHandlerChain implements MessageHandler, Ordered, BeanFactoryAware, BeanNameAware {
private static final String OUTPUT_CHANNEL_PROPERTY = "outputChannel";
@@ -73,6 +77,12 @@ public class MessageHandlerChain extends IntegrationObjectSupport implements Mes
private volatile int order = Ordered.LOWEST_PRECEDENCE;
private volatile String beanName;
private volatile BeanFactory beanFactory;
private volatile ChannelResolver channelResolver;
private volatile boolean initialized;
private final Object initializationMonitor = new Object();
@@ -94,7 +104,22 @@ public class MessageHandlerChain extends IntegrationObjectSupport implements Mes
return this.order;
}
public final void afterPropertiesSet() {
public void setBeanName(String beanName) {
this.beanName = beanName;
}
public void setBeanFactory(BeanFactory beanFactory) {
this.beanFactory = beanFactory;
}
private ChannelResolver getChannelResolver() {
if (this.channelResolver == null) {
this.channelResolver = new BeanFactoryChannelResolver(this.beanFactory);
}
return this.channelResolver;
}
private void initialize() {
synchronized (this.initializationMonitor) {
if (!this.initialized) {
Assert.notEmpty(this.handlers, "handler list must not be empty");
@@ -106,7 +131,7 @@ public class MessageHandlerChain extends IntegrationObjectSupport implements Mes
public void handleMessage(Message<?> message) {
if (!this.initialized) {
this.afterPropertiesSet();
this.initialize();
}
this.handlers.get(0).handleMessage(message);
}
@@ -148,7 +173,7 @@ public class MessageHandlerChain extends IntegrationObjectSupport implements Mes
private class ReplyForwardingMessageChannel implements MessageChannel {
public String getName() {
return MessageHandlerChain.this.getBeanName();
return MessageHandlerChain.this.beanName;
}
public boolean send(Message<?> message) {

18
org.springframework.integration/src/main/java/org/springframework/integration/handler/MethodInvokingMessageProcessor.java
View File

@@ -84,6 +84,8 @@ public class MethodInvokingMessageProcessor implements MessageProcessor {
private final Map<Class<?>, HandlerMethod> handlerMethods;
private volatile ConversionService conversionService;
private final EvaluationContext evaluationContext;
@@ -143,6 +145,17 @@ public class MethodInvokingMessageProcessor implements MessageProcessor {
this.displayString = sb.toString() + "]";
}
public void setConversionService(ConversionService conversionService) {
this.conversionService = conversionService;
}
private ConversionService getRequiredConversionService() {
if (this.conversionService == null) {
this.conversionService = ConversionServiceFactory.createDefaultConversionService();
}
return this.conversionService;
}
private EvaluationContext createEvaluationContext(Object method, Class<? extends Annotation> annotationType) {
StandardEvaluationContext context = new StandardEvaluationContext();
Class<?> targetType = AopUtils.getTargetClass(this.targetObject);
@@ -154,10 +167,7 @@ public class MethodInvokingMessageProcessor implements MessageProcessor {
new HandlerMethodFilter(annotationType, (String) method, this.requiresReply));
}
context.addPropertyAccessor(new MapAccessor());
// TODO: Enable configuration of an integration ConversionService bean to be used here,
// but then fallback to this same default if no such bean has been defined.
ConversionService conversionService = ConversionServiceFactory.createDefaultConversionService();
context.setTypeConverter(new StandardTypeConverter(conversionService));
context.setTypeConverter(new StandardTypeConverter(this.getRequiredConversionService()));
context.setVariable("target", targetObject);
return context;
}

4
org.springframework.integration/src/test/java/org/springframework/integration/channel/DispatchingChannelErrorHandlingTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2009 the original author or authors.
* Copyright 2002-2010 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.
@@ -68,6 +68,7 @@ public class DispatchingChannelErrorHandlingTests {
TaskExecutor executor = new SimpleAsyncTaskExecutor();
PublishSubscribeChannel channel = new PublishSubscribeChannel(executor);
channel.setBeanFactory(context);
channel.afterPropertiesSet();
ResultHandler resultHandler = new ResultHandler();
defaultErrorChannel.subscribe(resultHandler);
channel.subscribe(new MessageHandler() {
@@ -98,6 +99,7 @@ public class DispatchingChannelErrorHandlingTests {
TaskExecutor executor = new SimpleAsyncTaskExecutor();
ExecutorChannel channel = new ExecutorChannel(executor);
channel.setBeanFactory(context);
channel.afterPropertiesSet();
ResultHandler resultHandler = new ResultHandler();
defaultErrorChannel.subscribe(resultHandler);
channel.subscribe(new MessageHandler() {

1
org.springframework.integration/src/test/java/org/springframework/integration/gateway/GatewayProxyFactoryBeanTests.java
View File

@@ -266,6 +266,7 @@ public class GatewayProxyFactoryBeanTests {
proxyFactory.setBeanName("testGateway");
DirectChannel channel = new DirectChannel();
channel.setBeanName("testChannel");
channel.afterPropertiesSet();
EventDrivenConsumer consumer = new EventDrivenConsumer(channel, new BridgeHandler());
consumer.setBeanName("testBridge");
consumer.afterPropertiesSet();