diff --git a/spring-integration-core/src/main/java/org/springframework/integration/MessageSourceReceiveException.java b/spring-integration-core/src/main/java/org/springframework/integration/MessageSourceReceiveException.java new file mode 100644 index 0000000000..5a057672b1 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/MessageSourceReceiveException.java @@ -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; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/MessagePublishingErrorHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/MessagePublishingErrorHandler.java index e9a8cef828..c0156ccaa8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/MessagePublishingErrorHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/MessagePublishingErrorHandler.java @@ -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; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java index 888b37532d..55d7dd66f5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java @@ -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 source) { this.source = source; } @@ -120,6 +127,7 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean extends AbstractExpressionEvaluat @SuppressWarnings("unchecked") public final Message receive() { Message message = null; - Object result = this.doReceive(); - if (result == null) { - return null; - } + Object result = null; + Map 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) result; diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd index 5a53fa076e..148882ad92 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd @@ -685,6 +685,15 @@ + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/PollerWithErrorChannel-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/PollerWithErrorChannel-context.xml new file mode 100644 index 0000000000..93cc274774 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/PollerWithErrorChannel-context.xml @@ -0,0 +1,25 @@ + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/PollerWithErrorChannel.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/PollerWithErrorChannel.java new file mode 100644 index 0000000000..96459f6bf2 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/PollerWithErrorChannel.java @@ -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"; + } + } +}