Added namespace support for 'logging-channel-adapter' (INT-524). Moved default bean configuration into a BeanFactoryPostProcessor (INT-525). This also enables resolution of issue INT-520.
This commit is contained in:
Mark Fisher
@@ -86,7 +86,9 @@ public class MessagePublishingErrorHandler implements ErrorHandler, BeanFactoryA
|
||||
}
|
||||
}
|
||||
catch (Throwable errorDeliveryError) { // message will be logged only
|
||||
logger.error("Error message was not delivered, it will be dumped in the error log", errorDeliveryError);
|
||||
if (logger.isWarnEnabled()) {
|
||||
logger.warn("Error message was not delivered.", errorDeliveryError);
|
||||
}
|
||||
}
|
||||
}
|
||||
if (!sent && logger.isErrorEnabled()) {
|
||||
|
||||
@@ -0,0 +1,114 @@
|
||||
/*
|
||||
* 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.xml;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
|
||||
import org.springframework.beans.BeansException;
|
||||
import org.springframework.beans.factory.config.BeanDefinitionHolder;
|
||||
import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
|
||||
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
|
||||
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
|
||||
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
|
||||
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
|
||||
import org.springframework.beans.factory.support.RootBeanDefinition;
|
||||
import org.springframework.core.task.TaskExecutor;
|
||||
import org.springframework.integration.context.IntegrationContextUtils;
|
||||
|
||||
/**
|
||||
* A {@link BeanFactoryPostProcessor} implementation that provides default
|
||||
* beans for the error handling and task scheduling if those beans have not
|
||||
* already been explicitly defined within the registry.
|
||||
*
|
||||
* @author Mark Fisher
|
||||
*/
|
||||
class DefaultConfiguringBeanFactoryPostProcessor implements BeanFactoryPostProcessor {
|
||||
|
||||
private Log logger = LogFactory.getLog(this.getClass());
|
||||
|
||||
|
||||
public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
|
||||
if (beanFactory instanceof BeanDefinitionRegistry) {
|
||||
BeanDefinitionRegistry registry = (BeanDefinitionRegistry) beanFactory;
|
||||
this.registerErrorChannelIfNecessary(registry);
|
||||
this.registerTaskSchedulerIfNecessary(registry);
|
||||
}
|
||||
else if (logger.isWarnEnabled()) {
|
||||
logger.warn("BeanFactory is not a BeanDefinitionRegistry. The default '"
|
||||
+ IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME + "' and '"
|
||||
+ IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME + "' cannot be configured.");
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Register an error channel in the given BeanDefinitionRegistry if not yet present.
|
||||
* The bean name for which this is checking is defined by the constant
|
||||
* {@link IntegrationContextUtils#ERROR_CHANNEL_BEAN_NAME}.
|
||||
*/
|
||||
private void registerErrorChannelIfNecessary(BeanDefinitionRegistry registry) {
|
||||
if (!registry.isBeanNameInUse(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME)) {
|
||||
if (logger.isInfoEnabled()) {
|
||||
logger.info("No bean named '" + IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME +
|
||||
"' has been explicitly defined. Therefore, a default PublishSubscribeChannel will be created.");
|
||||
}
|
||||
RootBeanDefinition errorChannelDef = new RootBeanDefinition();
|
||||
errorChannelDef.setBeanClassName(IntegrationNamespaceUtils.BASE_PACKAGE + ".channel.PublishSubscribeChannel");
|
||||
BeanDefinitionHolder errorChannelHolder = new BeanDefinitionHolder(
|
||||
errorChannelDef, IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME);
|
||||
BeanDefinitionReaderUtils.registerBeanDefinition(errorChannelHolder, registry);
|
||||
BeanDefinitionBuilder loggingHandlerBuilder = BeanDefinitionBuilder.genericBeanDefinition(
|
||||
IntegrationNamespaceUtils.BASE_PACKAGE + ".handler.LoggingHandler");
|
||||
loggingHandlerBuilder.addConstructorArgValue("ERROR");
|
||||
String loggingHandlerBeanName = BeanDefinitionReaderUtils.registerWithGeneratedName(
|
||||
loggingHandlerBuilder.getBeanDefinition(), registry);
|
||||
BeanDefinitionBuilder loggingEndpointBuilder = BeanDefinitionBuilder.genericBeanDefinition(
|
||||
IntegrationNamespaceUtils.BASE_PACKAGE + ".endpoint.EventDrivenConsumer");
|
||||
loggingEndpointBuilder.addConstructorArgReference(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME);
|
||||
loggingEndpointBuilder.addConstructorArgReference(loggingHandlerBeanName);
|
||||
BeanDefinitionReaderUtils.registerWithGeneratedName(loggingEndpointBuilder.getBeanDefinition(), registry);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Register a TaskScheduler in the given BeanDefinitionRegistry if not yet present.
|
||||
* The bean name for which this is checking is defined by the constant
|
||||
* {@link IntegrationContextUtils#TASK_SCHEDULER_BEAN_NAME}.
|
||||
*/
|
||||
private void registerTaskSchedulerIfNecessary(BeanDefinitionRegistry registry) {
|
||||
if (!registry.isBeanNameInUse(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME)) {
|
||||
if (logger.isInfoEnabled()) {
|
||||
logger.info("No bean named '" + IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME +
|
||||
"' has been explicitly defined. Therefore, a default SimpleTaskScheduler will be created.");
|
||||
}
|
||||
TaskExecutor taskExecutor = IntegrationContextUtils.createThreadPoolTaskExecutor(2, 100, 0, "task-scheduler-");
|
||||
BeanDefinitionBuilder schedulerBuilder = BeanDefinitionBuilder.genericBeanDefinition(
|
||||
IntegrationNamespaceUtils.BASE_PACKAGE + ".scheduling.SimpleTaskScheduler");
|
||||
schedulerBuilder.addConstructorArgValue(taskExecutor);
|
||||
BeanDefinitionBuilder errorHandlerBuilder = BeanDefinitionBuilder.genericBeanDefinition(
|
||||
IntegrationNamespaceUtils.BASE_PACKAGE + ".channel.MessagePublishingErrorHandler");
|
||||
errorHandlerBuilder.addPropertyReference("defaultErrorChannel", IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME);
|
||||
String errorHandlerBeanName = BeanDefinitionReaderUtils.registerWithGeneratedName(
|
||||
errorHandlerBuilder.getBeanDefinition(), registry);
|
||||
schedulerBuilder.addPropertyReference("errorHandler", errorHandlerBeanName);
|
||||
BeanDefinitionHolder schedulerHolder = new BeanDefinitionHolder(
|
||||
schedulerBuilder.getBeanDefinition(), IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME);
|
||||
BeanDefinitionReaderUtils.registerBeanDefinition(schedulerHolder, registry);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -23,16 +23,9 @@ import org.springframework.beans.factory.config.BeanDefinition;
|
||||
import org.springframework.beans.factory.config.BeanDefinitionHolder;
|
||||
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
|
||||
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
|
||||
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
|
||||
import org.springframework.beans.factory.support.RootBeanDefinition;
|
||||
import org.springframework.beans.factory.xml.NamespaceHandler;
|
||||
import org.springframework.beans.factory.xml.NamespaceHandlerSupport;
|
||||
import org.springframework.beans.factory.xml.ParserContext;
|
||||
import org.springframework.core.task.TaskExecutor;
|
||||
import org.springframework.integration.channel.MessagePublishingErrorHandler;
|
||||
import org.springframework.integration.channel.PublishSubscribeChannel;
|
||||
import org.springframework.integration.context.IntegrationContextUtils;
|
||||
import org.springframework.integration.scheduling.SimpleTaskScheduler;
|
||||
|
||||
/**
|
||||
* Namespace handler for the integration namespace.
|
||||
@@ -42,52 +35,39 @@ import org.springframework.integration.scheduling.SimpleTaskScheduler;
|
||||
*/
|
||||
public class IntegrationNamespaceHandler implements NamespaceHandler {
|
||||
|
||||
private volatile boolean initializedContext;
|
||||
|
||||
private final NamespaceHandlerSupport delegate = new NamespaceHandlerDelegate();
|
||||
|
||||
private volatile boolean initialized;
|
||||
|
||||
private final Object initializationMonitor = new Object();
|
||||
|
||||
|
||||
public void init() {
|
||||
this.delegate.init();
|
||||
}
|
||||
|
||||
public BeanDefinition parse(Element element, ParserContext parserContext) {
|
||||
if (!this.initializedContext) {
|
||||
registerTaskSchedulerIfNecessary(parserContext.getRegistry());
|
||||
this.initializedContext = true;
|
||||
if (!this.initialized) {
|
||||
this.registerDefaultConfiguringBeanFactoryPostProcessor(parserContext);
|
||||
}
|
||||
return this.delegate.parse(element, parserContext);
|
||||
}
|
||||
}
|
||||
|
||||
public BeanDefinitionHolder decorate(Node source, BeanDefinitionHolder definition, ParserContext parserContext) {
|
||||
return this.delegate.decorate(source, definition, parserContext);
|
||||
}
|
||||
|
||||
/**
|
||||
* Register a TaskScheduler in the given BeanDefinitionRegistry if not yet present.
|
||||
* The bean name for which this is checking is defined by the constant
|
||||
* {@link IntegrationContextUtils#TASK_SCHEDULER_BEAN_NAME}.
|
||||
*/
|
||||
private static void registerTaskSchedulerIfNecessary(BeanDefinitionRegistry registry) {
|
||||
if (!registry.containsBeanDefinition(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME)) {
|
||||
RootBeanDefinition errorChannelDef = new RootBeanDefinition(PublishSubscribeChannel.class);
|
||||
BeanDefinitionHolder errorChannelHolder = new BeanDefinitionHolder(
|
||||
errorChannelDef, IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME);
|
||||
BeanDefinitionReaderUtils.registerBeanDefinition(errorChannelHolder, registry);
|
||||
}
|
||||
TaskExecutor taskExecutor = null;
|
||||
if (!registry.containsBeanDefinition(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME)) {
|
||||
taskExecutor = IntegrationContextUtils.createThreadPoolTaskExecutor(2, 100, 0, "task-scheduler-");
|
||||
BeanDefinitionBuilder schedulerBuilder = BeanDefinitionBuilder.genericBeanDefinition(SimpleTaskScheduler.class);
|
||||
schedulerBuilder.addConstructorArgValue(taskExecutor);
|
||||
BeanDefinitionBuilder errorHandlerBuilder = BeanDefinitionBuilder.genericBeanDefinition(MessagePublishingErrorHandler.class);
|
||||
errorHandlerBuilder.addPropertyReference("defaultErrorChannel", IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME);
|
||||
String errorHandlerBeanName = BeanDefinitionReaderUtils.registerWithGeneratedName(
|
||||
errorHandlerBuilder.getBeanDefinition(), registry);
|
||||
schedulerBuilder.addPropertyReference("errorHandler", errorHandlerBeanName);
|
||||
BeanDefinitionHolder schedulerHolder = new BeanDefinitionHolder(
|
||||
schedulerBuilder.getBeanDefinition(), IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME);
|
||||
BeanDefinitionReaderUtils.registerBeanDefinition(schedulerHolder, registry);
|
||||
private void registerDefaultConfiguringBeanFactoryPostProcessor(ParserContext parserContext) {
|
||||
synchronized (this.initializationMonitor) {
|
||||
if (!this.initialized) {
|
||||
String postProcessorSimpleClassName = "DefaultConfiguringBeanFactoryPostProcessor";
|
||||
BeanDefinitionBuilder postProcessorBuilder = BeanDefinitionBuilder.genericBeanDefinition(
|
||||
IntegrationNamespaceUtils.BASE_PACKAGE + ".config.xml." + postProcessorSimpleClassName);
|
||||
BeanDefinitionHolder postProcessorHolder = new BeanDefinitionHolder(postProcessorBuilder.getBeanDefinition(),
|
||||
IntegrationNamespaceUtils.BASE_PACKAGE + ".internal" + postProcessorSimpleClassName);
|
||||
BeanDefinitionReaderUtils.registerBeanDefinition(postProcessorHolder, parserContext.getRegistry());
|
||||
this.initialized = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -111,6 +91,7 @@ public class IntegrationNamespaceHandler implements NamespaceHandler {
|
||||
registerBeanDefinitionParser("payload-deserializing-transformer", new PayloadDeserializingTransformerParser());
|
||||
registerBeanDefinitionParser("inbound-channel-adapter", new MethodInvokingInboundChannelAdapterParser());
|
||||
registerBeanDefinitionParser("outbound-channel-adapter", new MethodInvokingOutboundChannelAdapterParser());
|
||||
registerBeanDefinitionParser("logging-channel-adapter", new LoggingChannelAdapterParser());
|
||||
registerBeanDefinitionParser("gateway", new GatewayParser());
|
||||
registerBeanDefinitionParser("bridge", new BridgeParser());
|
||||
registerBeanDefinitionParser("chain", new ChainParser());
|
||||
|
||||
@@ -182,7 +182,7 @@
|
||||
</xsd:complexType>
|
||||
</xsd:element>
|
||||
|
||||
<xsd:element name="inbound-channel-adapter" type="channelAdapterType">
|
||||
<xsd:element name="inbound-channel-adapter" type="methodInvokingChannelAdapterType">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation>
|
||||
Defines a Channel Adapter that receives from a MessageSource and sends to a MessageChannel.
|
||||
@@ -190,7 +190,7 @@
|
||||
</xsd:annotation>
|
||||
</xsd:element>
|
||||
|
||||
<xsd:element name="outbound-channel-adapter" type="channelAdapterType">
|
||||
<xsd:element name="outbound-channel-adapter" type="methodInvokingChannelAdapterType">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation>
|
||||
Defines a Channel Adapter that receives from a MessageChannel and sends to a MessageConsumer.
|
||||
@@ -198,6 +198,36 @@
|
||||
</xsd:annotation>
|
||||
</xsd:element>
|
||||
|
||||
<xsd:element name="logging-channel-adapter">
|
||||
<xsd:complexType>
|
||||
<xsd:complexContent>
|
||||
<xsd:extension base="channelAdapterType">
|
||||
<xsd:attribute name="level" default="INFO">
|
||||
<xsd:simpleType>
|
||||
<xsd:restriction base="xsd:string">
|
||||
<xsd:enumeration value="FATAL"/>
|
||||
<xsd:enumeration value="ERROR"/>
|
||||
<xsd:enumeration value="WARN"/>
|
||||
<xsd:enumeration value="INFO"/>
|
||||
<xsd:enumeration value="DEBUG"/>
|
||||
<xsd:enumeration value="TRACE"/>
|
||||
</xsd:restriction>
|
||||
</xsd:simpleType>
|
||||
</xsd:attribute>
|
||||
</xsd:extension>
|
||||
</xsd:complexContent>
|
||||
</xsd:complexType>
|
||||
</xsd:element>
|
||||
|
||||
<xsd:complexType name="methodInvokingChannelAdapterType">
|
||||
<xsd:complexContent>
|
||||
<xsd:extension base="channelAdapterType">
|
||||
<xsd:attribute name="ref" type="xsd:string" use="required"/>
|
||||
<xsd:attribute name="method" type="xsd:string"/>
|
||||
</xsd:extension>
|
||||
</xsd:complexContent>
|
||||
</xsd:complexType>
|
||||
|
||||
<xsd:complexType name="channelAdapterType">
|
||||
<xsd:sequence>
|
||||
<xsd:element name="poller" type="innerPollerType" minOccurs="0" maxOccurs="1"/>
|
||||
@@ -212,8 +242,6 @@
|
||||
</xsd:appinfo>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
<xsd:attribute name="ref" type="xsd:string" use="required"/>
|
||||
<xsd:attribute name="method" type="xsd:string"/>
|
||||
<xsd:attribute name="auto-startup" type="xsd:string" default="true"/>
|
||||
</xsd:complexType>
|
||||
|
||||
|
||||
Reference in New Issue
Block a user