diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/EnricherParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/EnricherParser.java index 32a7cdd873..0f97644fdb 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/EnricherParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/EnricherParser.java @@ -41,6 +41,7 @@ import org.springframework.util.xml.DomUtils; * @author Mark Fisher * @author Artem Bilan * @author Liujiong + * @author Kris Jacyna * @since 2.1 */ public class EnricherParser extends AbstractConsumerEndpointParser { @@ -51,6 +52,7 @@ public class EnricherParser extends AbstractConsumerEndpointParser { IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "request-channel"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-channel"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "request-timeout"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "reply-timeout"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "requires-reply"); @@ -69,19 +71,21 @@ public class EnricherParser extends AbstractConsumerEndpointParser { boolean hasAttributeExpression = StringUtils.hasText(expression); boolean hasAttributeNullResultExpression = StringUtils.hasText(nullResultExpression); - if (hasAttributeValue && hasAttributeExpression){ + if (hasAttributeValue && hasAttributeExpression) { parserContext.getReaderContext().error("Only one of 'value' or 'expression' is allowed", element); } - if (!hasAttributeValue && !hasAttributeExpression && !hasAttributeNullResultExpression){ - parserContext.getReaderContext().error("One of 'value' or 'expression' or 'null-result-expression' is required", element); + if (!hasAttributeValue && !hasAttributeExpression && !hasAttributeNullResultExpression) { + parserContext.getReaderContext() + .error("One of 'value' or 'expression' or 'null-result-expression' is required", element); } BeanDefinition expressionDef = null; BeanDefinition nullResultExpressionExpressionDef; if (hasAttributeValue) { - BeanDefinitionBuilder expressionBuilder = BeanDefinitionBuilder.genericBeanDefinition(ValueExpression.class); + BeanDefinitionBuilder expressionBuilder = + BeanDefinitionBuilder.genericBeanDefinition(ValueExpression.class); if (StringUtils.hasText(type)) { expressionBuilder.addConstructorArgValue(new TypedStringValue(value, type)); } @@ -93,19 +97,20 @@ public class EnricherParser extends AbstractConsumerEndpointParser { else if (hasAttributeExpression) { if (StringUtils.hasText(type)) { parserContext.getReaderContext().error("The 'type' attribute for '' of '' " + - "is not allowed with an 'expression' attribute.", element); + "is not allowed with an 'expression' attribute.", element); } expressionDef = BeanDefinitionBuilder .genericBeanDefinition(ExpressionFactoryBean.class) .addConstructorArgValue(expression) .getBeanDefinition(); } - if (expressionDef != null){ + if (expressionDef != null) { expressions.put(name, expressionDef); } if (hasAttributeNullResultExpression) { - nullResultExpressionExpressionDef = BeanDefinitionBuilder.genericBeanDefinition(ExpressionFactoryBean.class) - .addConstructorArgValue(nullResultExpression).getBeanDefinition(); + nullResultExpressionExpressionDef = + BeanDefinitionBuilder.genericBeanDefinition(ExpressionFactoryBean.class) + .addConstructorArgValue(nullResultExpression).getBeanDefinition(); nullResultExpressions.put(name, nullResultExpressionExpressionDef); } } @@ -129,13 +134,14 @@ public class EnricherParser extends AbstractConsumerEndpointParser { boolean hasAttributeValue = StringUtils.hasText(valueElementValue); boolean hasAttributeExpression = StringUtils.hasText(expressionElementValue); boolean hasAttributeNullResultExpression = StringUtils.hasText(nullResultHeaderExpression); - if (hasAttributeValue && hasAttributeExpression){ + if (hasAttributeValue && hasAttributeExpression) { parserContext.getReaderContext().error("Only one of '" + "value" + "' or '" - + "expression" + "' is allowed", subElement); + + "expression" + "' is allowed", subElement); } - if (!hasAttributeValue && !hasAttributeExpression && !hasAttributeNullResultExpression){ - parserContext.getReaderContext().error("One of 'value' or 'expression' or 'null-result-expression' is required", subElement); + if (!hasAttributeValue && !hasAttributeExpression && !hasAttributeNullResultExpression) { + parserContext.getReaderContext() + .error("One of 'value' or 'expression' or 'null-result-expression' is required", subElement); } BeanDefinition expressionDef = null; if (hasAttributeValue) { @@ -143,7 +149,8 @@ public class EnricherParser extends AbstractConsumerEndpointParser { expressionDef.getConstructorArgumentValues().addGenericArgumentValue(valueElementValue); } else if (hasAttributeExpression) { - expressionDef = IntegrationNamespaceUtils.createExpressionDefIfAttributeDefined("expression", subElement); + expressionDef = + IntegrationNamespaceUtils.createExpressionDefIfAttributeDefined("expression", subElement); } if (StringUtils.hasText(subElement.getAttribute("expression")) @@ -167,7 +174,8 @@ public class EnricherParser extends AbstractConsumerEndpointParser { .genericBeanDefinition(ExpressionEvaluatingHeaderValueMessageProcessor.class) .addConstructorArgValue(nullResultExpressionDefinition) .addConstructorArgValue(subElement.getAttribute("type")); - IntegrationNamespaceUtils.setValueIfAttributeDefined(nullResultValueProcessorBuilder, subElement, "overwrite"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(nullResultValueProcessorBuilder, subElement, + "overwrite"); nullResultHeaderExpressions.put(name, nullResultValueProcessorBuilder.getBeanDefinition()); } } @@ -184,8 +192,9 @@ public class EnricherParser extends AbstractConsumerEndpointParser { String requestPayloadExpression = element.getAttribute("request-payload-expression"); if (StringUtils.hasText(requestPayloadExpression)) { - BeanDefinitionBuilder expressionBuilder = BeanDefinitionBuilder.genericBeanDefinition(ExpressionFactoryBean.class); - expressionBuilder.addConstructorArgValue(requestPayloadExpression); + BeanDefinitionBuilder expressionBuilder = + BeanDefinitionBuilder.genericBeanDefinition(ExpressionFactoryBean.class) + .addConstructorArgValue(requestPayloadExpression); builder.addPropertyValue("requestPayloadExpression", expressionBuilder.getBeanDefinition()); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transformer/ContentEnricher.java b/spring-integration-core/src/main/java/org/springframework/integration/transformer/ContentEnricher.java index bab53b0dd8..646ed18418 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transformer/ContentEnricher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transformer/ContentEnricher.java @@ -49,6 +49,7 @@ import org.springframework.util.ReflectionUtils; * @author Gary Russell * @author Artem Bilan * @author Liujiong + * @author Kris Jacyna * @since 2.1 */ public class ContentEnricher extends AbstractReplyProducingMessageHandler @@ -82,6 +83,10 @@ public class ContentEnricher extends AbstractReplyProducingMessageHandler private volatile String replyChannelName; + private volatile MessageChannel errorChannel; + + private volatile String errorChannelName; + private volatile Gateway gateway = null; private volatile Long requestTimeout; @@ -167,6 +172,22 @@ public class ContentEnricher extends AbstractReplyProducingMessageHandler this.replyChannelName = replyChannelName; } + /** + * Set the content enricher's error channel to allow the error handling flow to return + * of an alternative object to use for enrichment if exceptions occur in the + * downstream flow. + * @param errorChannel The error channel. + * @since 4.1 + */ + public void setErrorChannel(MessageChannel errorChannel) { + this.errorChannel = errorChannel; + } + + public void setErrorChannelName(String errorChannelName) { + Assert.hasText(errorChannelName, "'errorChannelName' must not be empty"); + this.errorChannelName = errorChannelName; + } + /** * Set the timeout value for sending request messages. If not explicitly configured, * the default is one second. @@ -246,10 +267,17 @@ public class ContentEnricher extends AbstractReplyProducingMessageHandler Assert.state(!(this.replyChannelName != null && this.replyChannel != null), "'replyChannelName' and 'replyChannel' are mutually exclusive."); + Assert.state(!(this.errorChannelName != null && this.errorChannel != null), + "'errorChannelName' and 'errorChannel' are mutually exclusive."); + if (this.replyChannel != null || this.replyChannelName != null) { Assert.state(this.requestChannel != null || this.requestChannelName != null, "If the replyChannel is set, then the requestChannel must not be null"); } + if (this.errorChannel != null || this.errorChannelName != null) { + Assert.state(this.requestChannel != null || this.requestChannelName != null, + "If the errorChannel is set, then the requestChannel must not be null"); + } if (this.requestChannel != null || this.requestChannelName != null) { this.gateway = new Gateway(); this.gateway.setRequestChannel(this.requestChannel); @@ -269,6 +297,11 @@ public class ContentEnricher extends AbstractReplyProducingMessageHandler this.gateway.setReplyChannelName(this.replyChannelName); } + this.gateway.setErrorChannel(errorChannel); + if (this.errorChannelName != null) { + this.gateway.setErrorChannelName(this.errorChannelName); + } + if (this.getBeanFactory() != null) { this.gateway.setBeanFactory(this.getBeanFactory()); } diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd index 146c73cec4..0ce6fa47b2 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd @@ -1387,6 +1387,21 @@ + + + + Used when exceptions occur in a downstream flow and allows the error + handling flow to return an alternative object to use for enrichment. + If no "error-channel" reference is provided, + this enricher will propagate Exceptions to the caller. + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/EnricherParserTests5.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/EnricherParserTests5.java new file mode 100644 index 0000000000..ad0f17ab1f --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/EnricherParserTests5.java @@ -0,0 +1,97 @@ +/* + * Copyright 2014 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 org.junit.Assert.assertEquals; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.ApplicationContext; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; +import org.springframework.messaging.PollableChannel; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * Tests the error-channel in an enricher to produce + * a default object in case of downstream failure. + * + * @author Kris Jacyna + * @since 4.1 + */ +@RunWith(SpringJUnit4ClassRunner.class) +@ContextConfiguration +public class EnricherParserTests5 { + + @Autowired + private ApplicationContext context; + + + @Test + public void errorChannelTest() { + + class ErrorThrower extends AbstractReplyProducingMessageHandler { + @Override + protected Object handleRequestMessage(Message requestMessage) { + throw new RuntimeException(); + } + } + + class DefaultTargetProducer extends AbstractReplyProducingMessageHandler { + @Override + protected Object handleRequestMessage(Message requestMessage) { + final Target defaultTarget = new Target(); + defaultTarget.setName("Default"); + return defaultTarget; + } + } + + context.getBean("requestChannel", DirectChannel.class).subscribe(new ErrorThrower()); + context.getBean("errChannel", DirectChannel.class).subscribe(new DefaultTargetProducer()); + + Target original = new Target(); + original.setName("John"); + Message request = MessageBuilder.withPayload(original).build(); + + context.getBean("inputChannel", DirectChannel.class).send(request); + + Message reply = context.getBean("outputChannel", PollableChannel.class).receive(10000); + Target enriched = (Target) reply.getPayload(); + assertEquals("Mr. Default", enriched.getName()); + } + + + public static class Target { + + private volatile String name; + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/transformer/ContentEnricherTests.java b/spring-integration-core/src/test/java/org/springframework/integration/transformer/ContentEnricherTests.java index e63228c663..db38fe24fc 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/transformer/ContentEnricherTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/transformer/ContentEnricherTests.java @@ -57,6 +57,7 @@ import org.springframework.scheduling.support.PeriodicTrigger; * @author Gunnar Hillert * @author Artem Bilan * @author Gary Russell + * @author Kris Jacyna * * @since 2.1 */ @@ -85,9 +86,9 @@ public class ContentEnricherTests { public void replyChannelReplyTimingOut() throws Exception { final long requestTimeout = 500L; - final long replyTimeout = 700L; + final long replyTimeout = 700L; - final DirectChannel replyChannel = new DirectChannel(); + final DirectChannel replyChannel = new DirectChannel(); final QueueChannel requestChannel = new QueueChannel(1); final ContentEnricher enricher = new ContentEnricher(); @@ -144,10 +145,11 @@ public class ContentEnricherTests { Message requestMessage = MessageBuilder.withPayload(target).setReplyChannel(replyChannel).build(); try { - enricher.handleMessage(requestMessage); + enricher.handleMessage(requestMessage); } catch (ReplyRequiredException e) { - assertEquals("No reply produced by handler 'Enricher', and its 'requiresReply' property is set to true.", e.getMessage()); + assertEquals("No reply produced by handler 'Enricher', and its 'requiresReply' property is set to true.", + e.getMessage()); return; } @@ -161,7 +163,7 @@ public class ContentEnricherTests { final String requestChannelName = "Request_Channel"; final long requestTimeout = 200L; - QueueChannel replyChannel = new QueueChannel(); + QueueChannel replyChannel = new QueueChannel(); QueueChannel requestChannel = new RendezvousChannel(); requestChannel.setBeanName(requestChannelName); @@ -179,8 +181,7 @@ public class ContentEnricherTests { } catch (MessageDeliveryException e) { assertEquals("failed to send message to channel '" + requestChannelName - + "' within timeout: " + requestTimeout, e.getMessage()); - return; + + "' within timeout: " + requestTimeout, e.getMessage()); } } @@ -216,7 +217,7 @@ public class ContentEnricherTests { @Test public void setReplyChannelWithoutRequestChannel() { - QueueChannel replyChannel = new QueueChannel(); + QueueChannel replyChannel = new QueueChannel(); ContentEnricher enricher = new ContentEnricher(); enricher.setReplyChannel(replyChannel); @@ -287,8 +288,8 @@ public class ContentEnricherTests { @Test public void testContentEnricherWithNullRequestChannel() { - ContentEnricher enricher = new ContentEnricher(); - enricher.setReplyChannel(new QueueChannel()); + ContentEnricher enricher = new ContentEnricher(); + enricher.setReplyChannel(new QueueChannel()); enricher.setBeanFactory(mock(BeanFactory.class)); try { @@ -383,7 +384,7 @@ public class ContentEnricherTests { enricher.afterPropertiesSet(); TargetUser target = new TargetUser(); - target.setName("replace me"); + target.setName("replace me"); Message requestMessage = MessageBuilder.withPayload(target).setReplyChannel(replyChannel).build(); enricher.handleMessage(requestMessage); @@ -416,12 +417,12 @@ public class ContentEnricherTests { enricher.afterPropertiesSet(); UncloneableTargetUser target = new UncloneableTargetUser(); - target.setName("replace me"); + target.setName("replace me"); Message requestMessage = MessageBuilder.withPayload(target).setReplyChannel(replyChannel).build(); try { - enricher.handleMessage(requestMessage); + enricher.handleMessage(requestMessage); } catch (MessageHandlingException e) { assertThat(e.getMessage(), containsString("Failed to clone payload object")); @@ -469,6 +470,56 @@ public class ContentEnricherTests { assertTrue(enricher.isRunning()); } + /** + * In this test a {@link Target} message is passed into a {@link ContentEnricher}. + * The Enricher passes the message to a "request-channel" to a handler which throws + * an exception. The {@link ContentEnricher} then uses the error flow and consults + * the "error-channel" which returns a alternative {@link Target}. + */ + @Test + public void testErrorChannel() throws Exception { + + final DirectChannel requestChannel = new DirectChannel(); + requestChannel.subscribe(new AbstractReplyProducingMessageHandler() { + @Override + protected Object handleRequestMessage(Message requestMessage) { + throw new RuntimeException(); + } + + }); + + final DirectChannel errorChannel = new DirectChannel(); + errorChannel.subscribe(new AbstractReplyProducingMessageHandler() { + @Override + protected Object handleRequestMessage(Message requestMessage) { + return new Target("failed"); + } + + }); + + final QueueChannel replyChannel = new QueueChannel(); + + final ContentEnricher enricher = new ContentEnricher(); + enricher.setRequestChannel(requestChannel); + enricher.setErrorChannel(errorChannel); + + SpelExpressionParser parser = new SpelExpressionParser(); + Map propertyExpressions = new HashMap(); + propertyExpressions.put("name", parser.parseExpression("payload.name + ' target'")); + + enricher.setPropertyExpressions(propertyExpressions); + enricher.setBeanFactory(mock(BeanFactory.class)); + enricher.afterPropertiesSet(); + + final Target target = new Target("replace me"); + Message requestMessage = MessageBuilder.withPayload(target).setReplyChannel(replyChannel).build(); + + enricher.handleMessage(requestMessage); + Message reply = replyChannel.receive(10000); + Target result = (Target) reply.getPayload(); + assertEquals("failed target", result.getName()); + } + @SuppressWarnings("unused") private static final class Source { diff --git a/src/reference/docbook/content-enrichment.xml b/src/reference/docbook/content-enrichment.xml index b78a5e266a..7d00b7c43a 100644 --- a/src/reference/docbook/content-enrichment.xml +++ b/src/reference/docbook/content-enrichment.xml @@ -173,7 +173,7 @@ with id integrationHeaderChannelRegistry and configure the required default delay using a constructor argument (milliseconds). - + Since version 4.1, you can set a property removeOnGet to true on the <bean/> definition, and the mapping entry will be removed @@ -212,7 +212,7 @@ - ]]> @@ -295,12 +295,13 @@ output-channel="" ]]> ]]> ]]> ]]> ]]> ]]> ]]> - ]]> ]]> ]]> @@ -381,7 +382,17 @@ Optional. - + + + Channel to which an ErrorMessage will be sent if an + Exception occurs downstream of the + request-channel. This enables you to return an alternative object to use + for enrichment. This is optional; if it is not set then Exception + is thrown to the caller. + Optional. + + + Maximum amount of time in milliseconds to wait when sending a message to the channel, if such channel may block. @@ -400,7 +411,7 @@ Optional. - + Boolean value indicating whether any payload that implements Cloneable should be cloned @@ -411,14 +422,14 @@ Optional. - + Allows you to configure a Message Poller if this endpoint is a Polling Consumer. Optional. - + Each property sub-element provides the name of a property (via the mandatory name @@ -452,7 +463,7 @@ returns null, it will be evaluated and the output of the evaluation will be returned instead. - + Each header sub-element provides the name of a Message header (via the mandatory name diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index eededc769f..2b9ac647c3 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -131,10 +131,17 @@
Content Enricher Improvements - Add null-result-expression attribute, which will be evaluated and returned if <enricher> returns null. + An null-result-expression attribute has been added, which is evaluated and returned if + <enricher> returns null. It can be added in <header> and <property>. See for more information. + + An error-channel attribute has been added, which is used to handle an error flow + if Exception occurs downstream of the request-channel. This enable + you to return an alternative object to use for enrichment. + See for more information. +
Header Channel Registry