From aeaf7d6e899a6faa2a940aabf214a837507c4a13 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Mon, 17 Oct 2011 15:33:25 -0400 Subject: [PATCH] initial implementation added enricher parser addressed PR comments --- .../config/xml/EnricherParser.java | 82 +++++++++ .../xml/IntegrationNamespaceHandler.java | 3 +- .../transformer/ContentEnricher.java | 161 +++++++++++++++++ .../config/xml/spring-integration-2.1.xsd | 104 +++++++++++ .../xml/EnricherParserTests-context.xml | 23 +++ .../config/xml/EnricherParserTests.java | 145 +++++++++++++++ .../transformer/ContentEnricherTests.java | 169 ++++++++++++++++++ 7 files changed, 686 insertions(+), 1 deletion(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/config/xml/EnricherParser.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/transformer/ContentEnricher.java create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/config/xml/EnricherParserTests-context.xml create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/config/xml/EnricherParserTests.java create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/transformer/ContentEnricherTests.java 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 new file mode 100644 index 0000000000..075f1b0080 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/EnricherParser.java @@ -0,0 +1,82 @@ +/* + * 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 java.util.List; + +import org.w3c.dom.Element; + +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.support.ManagedMap; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.expression.common.LiteralExpression; +import org.springframework.integration.config.ExpressionFactoryBean; +import org.springframework.integration.transformer.ContentEnricher; +import org.springframework.util.CollectionUtils; +import org.springframework.util.StringUtils; +import org.springframework.util.xml.DomUtils; + +/** + * Parser for the 'enricher' element. + * + * @author Mark Fisher + * @since 2.1 + */ +public class EnricherParser extends AbstractConsumerEndpointParser { + + @Override + protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) { + final BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(ContentEnricher.class); + String requestChannel = element.getAttribute("request-channel"); + String replyChannel = element.getAttribute("reply-channel"); + builder.addConstructorArgReference(requestChannel); + if (StringUtils.hasText(replyChannel)) { + builder.addConstructorArgReference(replyChannel); + } + List propertyElements = DomUtils.getChildElementsByTagName(element, "property"); + if (!CollectionUtils.isEmpty(propertyElements)) { + ManagedMap propertyExpressions = new ManagedMap(); + for (Element propertyElement : propertyElements) { + String name = propertyElement.getAttribute("name"); + String value = propertyElement.getAttribute("value"); + String expression = propertyElement.getAttribute("expression"); + if (StringUtils.hasText(value) && StringUtils.hasText(expression)) { + parserContext.getReaderContext().error("The 'value' and 'expression' attributes are mutually exclusive on " + + "an element's sub-element.", parserContext.extractSource(propertyElement)); + } + if (StringUtils.hasText(value)) { + BeanDefinitionBuilder expressionBuilder = BeanDefinitionBuilder.genericBeanDefinition(LiteralExpression.class); + expressionBuilder.addConstructorArgValue(value); + propertyExpressions.put(name, expressionBuilder.getBeanDefinition()); + } + else if (StringUtils.hasText(expression)) { + BeanDefinitionBuilder expressionBuilder = BeanDefinitionBuilder.genericBeanDefinition(ExpressionFactoryBean.class); + expressionBuilder.addConstructorArgValue(expression); + propertyExpressions.put(name, expressionBuilder.getBeanDefinition()); + } + else { + parserContext.getReaderContext().error("Exactly one of 'value' or 'expression' attributes must be provided on " + + "an element's sub-element.", parserContext.extractSource(propertyElement)); + } + } + builder.addPropertyValue("propertyExpressions", propertyExpressions); + } + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "should-clone-payload"); + return builder; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceHandler.java index 65c2fa222a..25377b2b8e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * 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. @@ -31,6 +31,7 @@ public class IntegrationNamespaceHandler extends AbstractIntegrationNamespaceHan registerBeanDefinitionParser("publish-subscribe-channel", new PublishSubscribeChannelParser()); registerBeanDefinitionParser("service-activator", new ServiceActivatorParser()); registerBeanDefinitionParser("transformer", new TransformerParser()); + registerBeanDefinitionParser("enricher", new EnricherParser()); registerBeanDefinitionParser("filter", new FilterParser()); registerBeanDefinitionParser("router", new DefaultRouterParser()); registerBeanDefinitionParser("header-value-router", new HeaderValueRouterParser()); 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 new file mode 100644 index 0000000000..268d669816 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/transformer/ContentEnricher.java @@ -0,0 +1,161 @@ +/* + * 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.transformer; + +import java.lang.reflect.Method; +import java.util.HashMap; +import java.util.Map; + +import org.springframework.context.Lifecycle; +import org.springframework.expression.Expression; +import org.springframework.expression.spel.SpelParserConfiguration; +import org.springframework.expression.spel.standard.SpelExpressionParser; +import org.springframework.expression.spel.support.StandardEvaluationContext; +import org.springframework.integration.Message; +import org.springframework.integration.MessageChannel; +import org.springframework.integration.MessageHandlingException; +import org.springframework.integration.gateway.MessagingGatewaySupport; +import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; +import org.springframework.util.Assert; +import org.springframework.util.ReflectionUtils; + +/** + * Content Enricher is a Message Transformer that invokes any downstream message flow via + * its request channel and then applies values from the reply Message to the original payload. + * + * @author Mark Fisher + * @since 2.1 + */ +public class ContentEnricher extends AbstractReplyProducingMessageHandler implements Lifecycle { + + private final Map propertyExpressions = new HashMap(); + + private final Gateway gateway = new Gateway(); + + private final SpelExpressionParser parser = new SpelExpressionParser(new SpelParserConfiguration(true, true)); + + private final StandardEvaluationContext evaluationContext = new StandardEvaluationContext(); + + private volatile boolean shouldClonePayload = false; + + + /** + * Create a Content Enricher with the given request channel. An anonymous reply channel + * will be created for each request. + */ + public ContentEnricher(MessageChannel requestChannel) { + this(requestChannel, null); + } + + /** + * Create a Content Enricher with the given request and reply channels. + */ + public ContentEnricher(MessageChannel requestChannel, MessageChannel replyChannel) { + Assert.notNull(requestChannel, "requestChannel must not be null"); + this.gateway.setRequestChannel(requestChannel); + if (replyChannel != null) { + this.gateway.setReplyChannel(replyChannel); + } + } + + + /** + * Provide the map of expressions to evaluate when enriching the target payload. + * The keys should simply be property names, and the values should be Expressions + * that will evaluate against the reply Message as the root object. + */ + public void setPropertyExpressions(Map propertyExpressions) { + Assert.notEmpty(propertyExpressions, "propertyExpressions must not be empty"); + synchronized (this.propertyExpressions) { + this.propertyExpressions.clear(); + for (Map.Entry entry : propertyExpressions.entrySet()) { + String key = entry.getKey(); + Expression value = entry.getValue(); + Assert.notNull(key, "propertyExpressions key must not be null"); + Assert.notNull(value, "propertyExpressions value must not be null"); + this.propertyExpressions.put(parser.parseExpression(key), value); + } + } + } + + /** + * Specify whether to clone payload objects to create the target object. + * This is only applicable for payload types that implement Cloneable. + */ + public void setShouldClonePayload(boolean shouldClonePayload) { + this.shouldClonePayload = shouldClonePayload; + } + + @Override + public void onInit() { + super.onInit(); + this.gateway.afterPropertiesSet(); + } + + @Override + protected Object handleRequestMessage(Message requestMessage) { + Object targetPayload = requestMessage.getPayload(); + if (targetPayload instanceof Cloneable && this.shouldClonePayload) { + try { + Method cloneMethod = targetPayload.getClass().getMethod("clone", new Class[0]); + targetPayload = ReflectionUtils.invokeMethod(cloneMethod, targetPayload); + } + catch (Exception e) { + throw new MessageHandlingException(requestMessage, "Failed to clone payload object", e); + } + } + Message replyMessage = this.gateway.sendAndReceiveMessage(requestMessage); + for (Map.Entry entry : this.propertyExpressions.entrySet()) { + Expression propertyExpression = entry.getKey(); + Expression valueExpression = entry.getValue(); + Object value = valueExpression.getValue(this.evaluationContext, replyMessage); + propertyExpression.setValue(targetPayload, value); + } + return targetPayload; + } + + + /* + * Lifecycle implementation + */ + + public void start() { + this.gateway.start(); + } + + public void stop() { + this.gateway.stop(); + } + + public boolean isRunning() { + return this.gateway.isRunning(); + } + + + /** + * Internal gateway implementation for request/reply handling. + * Simply exposes the sendAndReceiveMessage method. + */ + private static final class Gateway extends MessagingGatewaySupport { + + @Override + protected Message sendAndReceiveMessage(Object object) { + return super.sendAndReceiveMessage(object); + } + } + +} diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.1.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.1.xsd index 49e876e88e..640129c9ec 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.1.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.1.xsd @@ -981,6 +981,109 @@ endpoint itself is a Polling Consumer for a channel with a queue. + + + + Defines an endpoint that passes a Message to its request-channel + and then expects a reply Message. The reply Message then becomes + the root object for evaluation of expressions to enriche the + target payload. + + + + + + + + + + + + + + + + + + Each property sub-element provides the name of a property (via the required 'name' attribute). + That property should be settable on the target payload instance. Exactly one of the 'value' + or 'expression' attributes must be provided as well. The former for a literal value to set, + and the latter for a SpEL expression to be evaluated. The root object of the evaluation + context is the Message that was returned from the flow initiated by this enricher. + + + + + + + + Channel to which a Message will be sent to get the data to use for enrichment. + + + + + + + + + + + + Channel where a reply Message is expected. This is optional; typically the auto-generated + temporary reply channel is sufficient. + + + + + + + + + + + + Boolean value indicating whether any payload that implements Cloneable should be cloned + prior to sending the Message to the request chanenl for acquiring the enriching data. + The cloned version would be used as the target payload for the ultimate reply. + Default is false. + + + + + + + + + + + + + + + + + + + + + + + + + + + + + @@ -1113,6 +1216,7 @@ endpoint itself is a Polling Consumer for a channel with a queue. + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/EnricherParserTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/EnricherParserTests-context.xml new file mode 100644 index 0000000000..bcd6c4d5b0 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/EnricherParserTests-context.xml @@ -0,0 +1,23 @@ + + + + + + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/EnricherParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/EnricherParserTests.java new file mode 100644 index 0000000000..0a30de97b7 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/EnricherParserTests.java @@ -0,0 +1,145 @@ +/* + * 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 org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotSame; + +import java.util.Map; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.DirectFieldAccessor; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.ApplicationContext; +import org.springframework.expression.Expression; +import org.springframework.integration.Message; +import org.springframework.integration.MessageChannel; +import org.springframework.integration.core.PollableChannel; +import org.springframework.integration.core.SubscribableChannel; +import org.springframework.integration.endpoint.EventDrivenConsumer; +import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.integration.transformer.ContentEnricher; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Mark Fisher + * @since 2.1 + */ +@RunWith(SpringJUnit4ClassRunner.class) +@ContextConfiguration +public class EnricherParserTests { + + @Autowired + private ApplicationContext context; + + + @Test + @SuppressWarnings("unchecked") + public void configurationCheck() { + Object endpoint = context.getBean("enricher"); + assertEquals(EventDrivenConsumer.class, endpoint.getClass()); + Object handler = TestUtils.getPropertyValue(endpoint, "handler"); + assertEquals(ContentEnricher.class, handler.getClass()); + ContentEnricher enricher = (ContentEnricher) handler; + assertEquals(99, enricher.getOrder()); + DirectFieldAccessor accessor = new DirectFieldAccessor(enricher); + assertEquals(context.getBean("output"), accessor.getPropertyValue("outputChannel")); + assertEquals(true, accessor.getPropertyValue("shouldClonePayload")); + Map propertyExpressions = (Map) accessor.getPropertyValue("propertyExpressions"); + for (Map.Entry e : propertyExpressions.entrySet()) { + if ("name".equals(e.getKey().getExpressionString())) { + assertEquals("payload.sourceName", e.getValue().getExpressionString()); + } + else if ("age".equals(e.getKey().getExpressionString())) { + assertEquals("42", e.getValue().getExpressionString()); + } + else { + throw new IllegalStateException("expected 'name' and 'age' only, not: " + e.getKey().getExpressionString()); + } + } + } + + @Test + public void integrationTest() { + SubscribableChannel requests = context.getBean("requests", SubscribableChannel.class); + requests.subscribe(new AbstractReplyProducingMessageHandler() { + @Override + protected Object handleRequestMessage(Message requestMessage) { + return new Source("foo"); + } + }); + Target original = new Target(); + Message request = MessageBuilder.withPayload(original).build(); + context.getBean("input", MessageChannel.class).send(request); + Message reply = context.getBean("output", PollableChannel.class).receive(0); + Target enriched = (Target) reply.getPayload(); + assertEquals("foo", enriched.getName()); + assertEquals(42, enriched.getAge()); + assertNotSame(original, enriched); + } + + + private static class Source { + + private final String sourceName; + + Source(String sourceName) { + this.sourceName = sourceName; + } + + @SuppressWarnings("unused") + public String getSourceName() { + return sourceName; + } + } + + + public static class Target implements Cloneable { + + private volatile String name; + + private volatile int age; + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + public int getAge() { + return age; + } + + public void setAge(int age) { + this.age = age; + } + + public Object clone() { + Target copy = new Target(); + copy.setName(this.name); + copy.setAge(this.age); + return copy; + } + } + +} 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 new file mode 100644 index 0000000000..99f319a767 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/transformer/ContentEnricherTests.java @@ -0,0 +1,169 @@ +/* + * 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.transformer; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotSame; + +import java.util.HashMap; +import java.util.Map; + +import org.junit.Test; +import org.springframework.expression.Expression; +import org.springframework.expression.spel.standard.SpelExpressionParser; +import org.springframework.integration.Message; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; +import org.springframework.integration.support.MessageBuilder; + +/** + * @author Mark Fisher + * @since 2.1 + */ +public class ContentEnricherTests { + + @Test + public void simpleProperty() { + QueueChannel replyChannel = new QueueChannel(); + DirectChannel requestChannel = new DirectChannel(); + requestChannel.subscribe(new AbstractReplyProducingMessageHandler() { + @Override + protected Object handleRequestMessage(Message requestMessage) { + return new Source("John", "Doe"); + } + }); + ContentEnricher enricher = new ContentEnricher(requestChannel); + SpelExpressionParser parser = new SpelExpressionParser(); + Map propertyExpressions = new HashMap(); + propertyExpressions.put("name", parser.parseExpression("payload.lastName + ', ' + payload.firstName")); + enricher.setPropertyExpressions(propertyExpressions); + Target target = new Target("replace me"); + Message requestMessage = MessageBuilder.withPayload(target).setReplyChannel(replyChannel).build(); + enricher.handleMessage(requestMessage); + Message reply = replyChannel.receive(0); + assertEquals("Doe, John", ((Target) reply.getPayload()).getName()); + } + + @Test + public void nestedProperty() { + QueueChannel replyChannel = new QueueChannel(); + DirectChannel requestChannel = new DirectChannel(); + requestChannel.subscribe(new AbstractReplyProducingMessageHandler() { + @Override + protected Object handleRequestMessage(Message requestMessage) { + return new Source("John", "Doe"); + } + }); + ContentEnricher enricher = new ContentEnricher(requestChannel); + SpelExpressionParser parser = new SpelExpressionParser(); + Map propertyExpressions = new HashMap(); + propertyExpressions.put("child.name", parser.parseExpression("payload.lastName + ', ' + payload.firstName")); + enricher.setPropertyExpressions(propertyExpressions); + Target target = new Target("test"); + Message requestMessage = MessageBuilder.withPayload(target).setReplyChannel(replyChannel).build(); + enricher.handleMessage(requestMessage); + Message reply = replyChannel.receive(0); + Target result = (Target) reply.getPayload(); + assertEquals("test", result.getName()); + assertEquals("Doe, John", result.getChild().getName()); + } + + @Test + public void clonePayload() { + QueueChannel replyChannel = new QueueChannel(); + DirectChannel requestChannel = new DirectChannel(); + requestChannel.subscribe(new AbstractReplyProducingMessageHandler() { + @Override + protected Object handleRequestMessage(Message requestMessage) { + return new Source("John", "Doe"); + } + }); + ContentEnricher enricher = new ContentEnricher(requestChannel); + enricher.setShouldClonePayload(true); + SpelExpressionParser parser = new SpelExpressionParser(); + Map propertyExpressions = new HashMap(); + propertyExpressions.put("name", parser.parseExpression("payload.lastName + ', ' + payload.firstName")); + enricher.setPropertyExpressions(propertyExpressions); + Target target = new Target("replace me"); + Message requestMessage = MessageBuilder.withPayload(target).setReplyChannel(replyChannel).build(); + enricher.handleMessage(requestMessage); + Message reply = replyChannel.receive(0); + Target result = (Target) reply.getPayload(); + assertEquals("Doe, John", result.getName()); + assertNotSame(target, result); + } + + + @SuppressWarnings("unused") + private static final class Source { + + private final String firstName, lastName; + + Source(String firstName, String lastName) { + this.firstName = firstName; + this.lastName = lastName; + } + + public String getFirstName() { + return firstName; + } + + public String getLastName() { + return lastName; + } + } + + + public static final class Target implements Cloneable { + + private volatile String name; + + private volatile Target child; + + public Target() { + this.name = "default"; + } + + private Target(String name) { + this.name = name; + } + + public String getName() { + return this.name; + } + + public void setName(String name) { + this.name = name; + } + + public Target getChild() { + return this.child; + } + + public void setChild(Target child) { + this.child = child; + } + + public Object clone() { + Target clone = new Target(this.name); + clone.setChild(this.child); + return clone; + } + } + +} \ No newline at end of file