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

INT-1708, INT-1709 added initial support for 'error-channel' to inbound-channel-adapters

Browse Source
This commit is contained in:
Oleg Zhurakousky
2011-01-11 19:19:11 -05:00
parent 8630a10d23
commit 4bd0cb2c4b
8 changed files with 192 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
spring-integration-core/src/main/java/org/springframework/integration/MessageSourceReceiveException.java Normal file
View File

@@ -0,0 +1,40 @@
/*
* Copyright 2002-2011 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;
/**
* @author Oleg Zhurakousky
* @since 2.0.2
*/
@SuppressWarnings("serial")
public class MessageSourceReceiveException extends MessageHandlingException {
private volatile Object errorChannel;
public void setErrorChannel(Object errorChannel) {
this.errorChannel = errorChannel;
}
public MessageSourceReceiveException(Object errorChannel, Throwable t) {
super(null, t);
this.errorChannel = errorChannel;
}
public Object getErrorChannel() {
return errorChannel;
}
}

29
spring-integration-core/src/main/java/org/springframework/integration/channel/MessagePublishingErrorHandler.java
View File

@@ -18,10 +18,12 @@ package org.springframework.integration.channel;
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.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.MessageSourceReceiveException;
import org.springframework.integration.MessagingException;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.message.ErrorMessage;
@@ -29,6 +31,7 @@ import org.springframework.integration.support.channel.BeanFactoryChannelResolve
import org.springframework.integration.support.channel.ChannelResolver;
import org.springframework.util.Assert;
import org.springframework.util.ErrorHandler;
import org.springframework.util.StringUtils;
/**
* {@link ErrorHandler} implementation that sends an {@link ErrorMessage} to a
@@ -36,6 +39,7 @@ import org.springframework.util.ErrorHandler;
*
* @author Mark Fisher
* @author Iwein Fuld
* @author Oleg Zhurakousky
*/
public class MessagePublishingErrorHandler implements ErrorHandler, BeanFactoryAware {
@@ -73,9 +77,7 @@ public class MessagePublishingErrorHandler implements ErrorHandler, BeanFactoryA
}
public final void handleError(Throwable t) {
Message<?> failedMessage = (t instanceof MessagingException) ?
((MessagingException) t).getFailedMessage() : null;
MessageChannel errorChannel = this.resolveErrorChannel(failedMessage);
MessageChannel errorChannel = this.resolveErrorChannel(t);
boolean sent = false;
if (errorChannel != null) {
try {
@@ -93,6 +95,8 @@ public class MessagePublishingErrorHandler implements ErrorHandler, BeanFactoryA
}
}
if (!sent && logger.isErrorEnabled()) {
Message<?> failedMessage = (t instanceof MessagingException) ?
((MessagingException) t).getFailedMessage() : null;
if (failedMessage != null) {
logger.error("failure occurred in messaging task with message: " + failedMessage, t);
}
@@ -102,11 +106,28 @@ public class MessagePublishingErrorHandler implements ErrorHandler, BeanFactoryA
}
}
private MessageChannel resolveErrorChannel(Message<?> failedMessage) {
private MessageChannel resolveErrorChannel(Throwable t) {
Message<?> failedMessage = (t instanceof MessagingException) ?
((MessagingException) t).getFailedMessage() : null;
if (this.defaultErrorChannel == null && this.channelResolver != null) {
this.defaultErrorChannel = this.channelResolver.resolveChannelName(
IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME);
}
if (t instanceof MessageSourceReceiveException){
Object errorChannel = ((MessageSourceReceiveException)t).getErrorChannel();
if (errorChannel != null){
if (errorChannel instanceof MessageChannel){
return (MessageChannel) errorChannel;
}
else if (errorChannel instanceof String && StringUtils.hasText((String)errorChannel)){
return this.channelResolver.resolveChannelName((String) errorChannel);
}
else {
throw new MessagingException("Failed to resolve 'errorChannel' - " + errorChannel);
}
}
}
if (failedMessage == null || failedMessage.getHeaders().getErrorChannel() == null) {
return this.defaultErrorChannel;
}

8
spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java
View File

@@ -30,6 +30,7 @@ import org.springframework.integration.core.MessageSource;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.util.Assert;
import org.springframework.util.ErrorHandler;
/**
* FactoryBean for creating a SourcePollingChannelAdapter instance.
@@ -59,8 +60,14 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean<Sourc
private volatile boolean initialized;
private final Object initializationMonitor = new Object();
private volatile ErrorHandler errorHandler;
public void setErrorHandler(ErrorHandler errorHandler) {
this.errorHandler = errorHandler;
}
public void setSource(MessageSource<?> source) {
this.source = source;
}
@@ -120,6 +127,7 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean<Sourc
SourcePollingChannelAdapter spca = new SourcePollingChannelAdapter();
spca.setSource(this.source);
spca.setOutputChannel(this.outputChannel);
spca.setErrorHandler(errorHandler);
if (this.pollerMetadata == null) {
this.pollerMetadata = IntegrationContextUtils.getDefaultPollerMetadata(this.beanFactory);
Assert.notNull(this.pollerMetadata, "No poller has been defined for channel-adapter '"

10
spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractPollingInboundChannelAdapterParser.java
View File

@@ -22,6 +22,8 @@ import org.springframework.beans.BeanMetadataElement;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.channel.MessagePublishingErrorHandler;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
/**
@@ -46,6 +48,14 @@ public abstract class AbstractPollingInboundChannelAdapterParser extends Abstrac
IntegrationNamespaceUtils.configurePollerMetadata(pollerElement, adapterBuilder, parserContext);
}
IntegrationNamespaceUtils.setValueIfAttributeDefined(adapterBuilder, element, "auto-startup");
String errorChannel = element.getAttribute("error-channel");
if (StringUtils.hasText(errorChannel)){
BeanDefinitionBuilder errorHandler = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.integration.channel.MessagePublishingErrorHandler");
errorHandler.addPropertyReference("defaultErrorChannel", errorChannel);
adapterBuilder.addPropertyValue("errorHandler", errorHandler.getBeanDefinition());
}
return adapterBuilder.getBeanDefinition();
}

21
spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractMessageSource.java
View File

@@ -22,6 +22,8 @@ import java.util.Map;
import org.springframework.expression.Expression;
import org.springframework.integration.Message;
import org.springframework.integration.MessageHeaders;
import org.springframework.integration.MessageSourceReceiveException;
import org.springframework.integration.MessagingException;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.support.MessageBuilder;
@@ -45,11 +47,22 @@ public abstract class AbstractMessageSource<T> extends AbstractExpressionEvaluat
@SuppressWarnings("unchecked")
public final Message<T> receive() {
Message<T> message = null;
Object result = this.doReceive();
if (result == null) {
return null;
}
Object result = null;
Map<String, Object> headers = this.evaluateHeaders();
if (headers.containsKey(MessageHeaders.ERROR_CHANNEL)){
try {
result = this.doReceive();
}
catch (Exception e) {
throw new MessageSourceReceiveException((String) headers.get(MessageHeaders.ERROR_CHANNEL), e);
}
}
else {
result = this.doReceive();
}
if (result instanceof Message<?>) {
try {
message = (Message<T>) result;

9
spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd
View File

@@ -685,6 +685,15 @@
</xsd:sequence>
<xsd:attributeGroup ref="methodInvokingOrExpressionEvaluatingAttributes" />
<xsd:attributeGroup ref="channelAdapterAttributes" />
<xsd:attribute name="error-channel" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Identifies channel that error messages will be sent to if a failure occurs in this
adapter's invocation. To completely suppress Exceptions, provide a
reference to the "nullChannel" here.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>

25
spring-integration-core/src/test/java/org/springframework/integration/config/xml/PollerWithErrorChannel-context.xml Normal file
View File

@@ -0,0 +1,25 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd
http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.0.xsd"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:task="http://www.springframework.org/schema/task">
<int:inbound-channel-adapter id="withErrorHeader" expression="5/0" channel="serviceChannel" auto-startup="false">
<int:poller fixed-rate="1000"/>
<int:header name="errorChannel" value="eChannel"/>
</int:inbound-channel-adapter>
<int:inbound-channel-adapter id="withErrorChannel" expression="5/0"
channel="serviceChannel" auto-startup="false" error-channel="eChannel">
<int:poller fixed-rate="1000"/>
</int:inbound-channel-adapter>
<int:channel id="serviceChannel"/>
<int:channel id="eChannel">
<int:queue/>
</int:channel>
</beans>

58
spring-integration-core/src/test/java/org/springframework/integration/config/xml/PollerWithErrorChannel.java Normal file
View File

@@ -0,0 +1,58 @@
/*
* Copyright 2002-2011 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 static junit.framework.Assert.assertNotNull;
import org.junit.Test;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.core.PollableChannel;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
/**
* @author Oleg Zhurakousky
*
*/
public class PollerWithErrorChannel {
@Test
public void testWithErrorChannelAsHeader() throws Exception{
ApplicationContext ac = new ClassPathXmlApplicationContext("PollerWithErrorChannel-context.xml", this.getClass());
SourcePollingChannelAdapter adapter = ac.getBean("withErrorHeader", SourcePollingChannelAdapter.class);
adapter.start();
PollableChannel errorChannel = ac.getBean("eChannel", PollableChannel.class);
assertNotNull(errorChannel.receive(100000));
adapter.stop();
}
@Test
public void testWithErrorChannel() throws Exception{
ApplicationContext ac = new ClassPathXmlApplicationContext("PollerWithErrorChannel-context.xml", this.getClass());
SourcePollingChannelAdapter adapter = ac.getBean("withErrorChannel", SourcePollingChannelAdapter.class);
adapter.start();
PollableChannel errorChannel = ac.getBean("eChannel", PollableChannel.class);
assertNotNull(errorChannel.receive(100000));
adapter.stop();
}
public static class SampleService{
public String withSuccess(){
return "hello";
}
}
}