diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ScheduledProducerParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ScheduledProducerParser.java index 77b095a3a1..76de916cb3 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ScheduledProducerParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ScheduledProducerParser.java @@ -21,9 +21,9 @@ import java.util.List; import org.w3c.dom.Element; import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.support.BeanDefinitionReaderUtils; import org.springframework.beans.factory.support.ManagedMap; import org.springframework.beans.factory.support.RootBeanDefinition; -import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.util.CollectionUtils; import org.springframework.util.StringUtils; @@ -35,59 +35,22 @@ import org.springframework.util.xml.DomUtils; * @author Mark Fisher * @since 2.0 */ -public class ScheduledProducerParser extends AbstractSingleBeanDefinitionParser { +public class ScheduledProducerParser extends AbstractPollingInboundChannelAdapterParser { @Override - protected String getBeanClassName(Element element) { - return IntegrationNamespaceUtils.BASE_PACKAGE + ".endpoint.ScheduledMessageProducer"; + protected String parseSource(Element element, ParserContext parserContext) { + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition( + "org.springframework.integration.endpoint.ExpressionEvaluatingMessageSource"); + String payloadExpression = element.getAttribute("payload-expression"); + RootBeanDefinition expressionDef = new RootBeanDefinition("org.springframework.integration.config.ExpressionFactoryBean"); + expressionDef.getConstructorArgumentValues().addGenericArgumentValue(payloadExpression); + builder.addConstructorArgValue(expressionDef); + builder.addConstructorArgValue(null); // TODO: add support for expectedType? + this.parseHeaderExpressions(builder, element, parserContext); + return BeanDefinitionReaderUtils.registerWithGeneratedName(builder.getBeanDefinition(), parserContext.getRegistry()); } - @Override - protected boolean shouldGenerateIdAsFallback() { - return true; - } - - @Override - protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { - String fixedDelay = element.getAttribute("fixed-delay"); - String fixedRate = element.getAttribute("fixed-rate"); - String cron = element.getAttribute("cron"); - String trigger = element.getAttribute("trigger"); - int numTriggers = 0; - if (StringUtils.hasText(fixedDelay)) { - RootBeanDefinition triggerDefinition = new RootBeanDefinition( - "org.springframework.scheduling.support.PeriodicTrigger"); - triggerDefinition.getConstructorArgumentValues().addGenericArgumentValue(fixedDelay); - builder.addConstructorArgValue(triggerDefinition); - numTriggers++; - } - if (StringUtils.hasText(fixedRate)) { - RootBeanDefinition triggerDefinition = new RootBeanDefinition( - "org.springframework.scheduling.support.PeriodicTrigger"); - triggerDefinition.getConstructorArgumentValues().addGenericArgumentValue(fixedRate); - triggerDefinition.getPropertyValues().add("fixedRate", Boolean.TRUE); - builder.addConstructorArgValue(triggerDefinition); - numTriggers++; - } - if (StringUtils.hasText(cron)) { - RootBeanDefinition triggerDefinition = new RootBeanDefinition( - "org.springframework.scheduling.support.CronTrigger"); - triggerDefinition.getConstructorArgumentValues().addGenericArgumentValue(cron); - builder.addConstructorArgValue(triggerDefinition); - numTriggers++; - } - if (StringUtils.hasText(trigger)) { - builder.addConstructorArgReference(trigger); - numTriggers++; - } - if (numTriggers != 1) { - parserContext.getReaderContext().error("exactly one of the following trigger attributes must be provided: " - + "fixed-delay, fixed-rate, cron, or trigger", parserContext.extractSource(element)); - return; - } - builder.addPropertyReference("outputChannel", element.getAttribute("channel")); - builder.addConstructorArgValue(element.getAttribute("payload-expression")); - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup"); + private void parseHeaderExpressions(BeanDefinitionBuilder builder, Element element, ParserContext parserContext) { List headerElements = DomUtils.getChildElementsByTagName(element, "header"); if (!CollectionUtils.isEmpty(headerElements)) { ManagedMap headerExpressions = new ManagedMap(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractMessageSource.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractMessageSource.java new file mode 100644 index 0000000000..5ee0b4ec32 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractMessageSource.java @@ -0,0 +1,101 @@ +/* + * Copyright 2002-2010 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.endpoint; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +import org.springframework.expression.Expression; +import org.springframework.integration.Message; +import org.springframework.integration.MessagingException; +import org.springframework.integration.core.MessageSource; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.util.AbstractExpressionEvaluator; +import org.springframework.util.CollectionUtils; + +/** + * @author Mark Fisher + * @since 2.0 + */ +public abstract class AbstractMessageSource extends AbstractExpressionEvaluator implements MessageSource { + + private volatile Map headerExpressions = Collections.emptyMap(); + + + public void setHeaderExpressions(Map headerExpressions) { + this.headerExpressions = (headerExpressions != null) + ? headerExpressions : Collections.emptyMap(); + } + + @SuppressWarnings("unchecked") + public final Message receive() { + Message message = null; + Object result = this.doReceive(); + if (result == null) { + return null; + } + Map headers = this.evaluateHeaders(); + if (result instanceof Message) { + try { + message = (Message) result; + } + catch (Exception e) { + throw new MessagingException("MessageSource returned unexpected type.", e); + } + if (!CollectionUtils.isEmpty(headers)) { + // create a new Message from this one in order to apply headers + MessageBuilder builder = MessageBuilder.fromMessage(message); + builder.copyHeaders(headers); + message = builder.build(); + } + } + else { + T payload = null; + try { + payload = (T) result; + } + catch (Exception e) { + throw new MessagingException("MessageSource returned unexpected type.", e); + } + MessageBuilder builder = MessageBuilder.withPayload(payload); + if (!CollectionUtils.isEmpty(headers)) { + builder.copyHeaders(headers); + } + message = builder.build(); + } + return message; + } + + private Map evaluateHeaders() { + Map results = new HashMap(); + for (Map.Entry entry : this.headerExpressions.entrySet()) { + Object headerValue = this.evaluateExpression(entry.getValue()); + if (headerValue != null) { + results.put(entry.getKey(), headerValue); + } + } + return results; + } + + /** + * Subclasses must implement this method. Typically the returned value will be the payload of + * type T, but the returned value may also be a Message instance whose payload is of type T. + */ + protected abstract Object doReceive(); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ExpressionEvaluatingMessageSource.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ExpressionEvaluatingMessageSource.java new file mode 100644 index 0000000000..47aeb7bde2 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ExpressionEvaluatingMessageSource.java @@ -0,0 +1,43 @@ +/* + * Copyright 2002-2010 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.endpoint; + +import org.springframework.expression.Expression; +import org.springframework.util.Assert; + +/** + * @author Mark Fisher + * @since 2.0 + */ +public class ExpressionEvaluatingMessageSource extends AbstractMessageSource { + + private final Expression expression; + + private final Class expectedType; + + + public ExpressionEvaluatingMessageSource(Expression expression, Class expectedType) { + Assert.notNull(expression, "expression must not be null"); + this.expression = expression; + this.expectedType = expectedType; + } + + public T doReceive() { + return this.evaluateExpression(this.expression, this.expectedType); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ScheduledMessageProducer.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ScheduledMessageProducer.java deleted file mode 100644 index c069182774..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ScheduledMessageProducer.java +++ /dev/null @@ -1,124 +0,0 @@ -/* - * Copyright 2002-2010 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.endpoint; - -import java.util.HashMap; -import java.util.Map; -import java.util.concurrent.ScheduledFuture; - -import org.springframework.beans.factory.BeanFactory; -import org.springframework.expression.Expression; -import org.springframework.expression.ExpressionParser; -import org.springframework.expression.spel.standard.SpelExpressionParser; -import org.springframework.expression.spel.support.StandardEvaluationContext; -import org.springframework.integration.support.MessageBuilder; -import org.springframework.integration.util.SimpleBeanResolver; -import org.springframework.scheduling.Trigger; -import org.springframework.util.Assert; -import org.springframework.util.CollectionUtils; - -/** - * @author Mark Fisher - * @since 2.0 - */ -public class ScheduledMessageProducer extends MessageProducerSupport { - - private static final ExpressionParser PARSER = new SpelExpressionParser(); - - - private final Trigger trigger; - - private final MessageProducingTask task; - - private volatile ScheduledFuture future; - - private final Map headerExpressions = new HashMap(); - - private final StandardEvaluationContext context = new StandardEvaluationContext(); - - - public ScheduledMessageProducer(Trigger trigger, String payloadExpression) { - Assert.notNull(trigger, "trigger must not be null"); - Assert.hasText(payloadExpression, "payloadExpression is required"); - this.trigger = trigger; - this.task = new MessageProducingTask(PARSER.parseExpression(payloadExpression)); - } - - - public void setHeaderExpressions(Map headerExpressions) { - synchronized (this.headerExpressions) { - this.headerExpressions.clear(); - if (headerExpressions != null) { - this.headerExpressions.putAll(headerExpressions); - } - } - } - - private Map evaluateHeaders() { - Map headers = new HashMap(); - for (Map.Entry entry : this.headerExpressions.entrySet()) { - headers.put(entry.getKey(), entry.getValue().getValue(context)); - } - return headers; - } - - @Override - protected void onInit() { - super.onInit(); - final BeanFactory beanFactory = this.getBeanFactory(); - if (beanFactory != null) { - this.context.setBeanResolver(new SimpleBeanResolver(beanFactory)); - } - } - - @Override - protected void doStart() { - this.future = this.getTaskScheduler().schedule(this.task, this.trigger); - } - - @Override - protected void doStop() { - if (this.future != null) { - this.future.cancel(true); - } - } - - - private class MessageProducingTask implements Runnable { - - private final Expression payloadExpression; - - - private MessageProducingTask(Expression payloadExpression) {//, Map headerExpressions) { - this.payloadExpression = payloadExpression; - } - - - public void run() { - Object payload = this.payloadExpression.getValue(context); - if (payload != null) { - Map headers = evaluateHeaders(); - MessageBuilder builder = MessageBuilder.withPayload(payload); - if (!CollectionUtils.isEmpty(headers)) { - builder.copyHeaders(headers); - } - sendMessage(builder.build()); - } - } - } - -} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/AbstractExpressionEvaluator.java b/spring-integration-core/src/main/java/org/springframework/integration/util/AbstractExpressionEvaluator.java index f6ec55f759..3c13569d90 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/util/AbstractExpressionEvaluator.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/util/AbstractExpressionEvaluator.java @@ -97,6 +97,14 @@ public abstract class AbstractExpressionEvaluator implements BeanFactoryAware { return this.evaluateExpression(expression, input, (Class) null); } + protected T evaluateExpression(Expression expression, Class expectedType) { + return expression.getValue(this.evaluationContext, expectedType); + } + + protected Object evaluateExpression(Expression expression) { + return expression.getValue(this.evaluationContext); + } + protected T evaluateExpression(Expression expression, Object input, Class expectedType) { return expression.getValue(this.evaluationContext, input, expectedType); } 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 76ac85dfb3..2e95e67a30 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 @@ -2424,6 +2424,7 @@ Name of the header whose value to use. + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/ScheduledProducerParserTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/ScheduledProducerParserTests-context.xml index 71c0187ea6..c024d187a8 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/ScheduledProducerParserTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/ScheduledProducerParserTests-context.xml @@ -19,18 +19,27 @@ - + + + - + + + - + + + - + +
- + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/ScheduledProducerParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/ScheduledProducerParserTests.java index 048b7dbb58..203f34fe5a 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/ScheduledProducerParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/ScheduledProducerParserTests.java @@ -24,12 +24,12 @@ 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.endpoint.ScheduledMessageProducer; +import org.springframework.integration.endpoint.SourcePollingChannelAdapter; +import org.springframework.integration.test.util.TestUtils; import org.springframework.scheduling.Trigger; import org.springframework.scheduling.support.CronTrigger; import org.springframework.scheduling.support.PeriodicTrigger; @@ -50,71 +50,66 @@ public class ScheduledProducerParserTests { @Test public void fixedDelay() { - ScheduledMessageProducer producer = context.getBean("fixedDelayProducer", ScheduledMessageProducer.class); - assertFalse(producer.isAutoStartup()); - DirectFieldAccessor producerAccessor = new DirectFieldAccessor(producer); - Trigger trigger = (Trigger) producerAccessor.getPropertyValue("trigger"); + SourcePollingChannelAdapter adapter = context.getBean("fixedDelayProducer", SourcePollingChannelAdapter.class); + assertFalse(adapter.isAutoStartup()); + DirectFieldAccessor adapterAccessor = new DirectFieldAccessor(adapter); + Trigger trigger = TestUtils.getPropertyValue(adapter, "pollerMetadata.trigger", Trigger.class); assertEquals(PeriodicTrigger.class, trigger.getClass()); DirectFieldAccessor triggerAccessor = new DirectFieldAccessor(trigger); assertEquals(1234L, triggerAccessor.getPropertyValue("period")); assertEquals(Boolean.FALSE, triggerAccessor.getPropertyValue("fixedRate")); - assertEquals(context.getBean("fixedDelayChannel"), producerAccessor.getPropertyValue("outputChannel")); - Expression payloadExpression = (Expression) new DirectFieldAccessor( - producerAccessor.getPropertyValue("task")).getPropertyValue("payloadExpression"); - assertEquals("'fixedDelayTest'", payloadExpression.getExpressionString()); + assertEquals(context.getBean("fixedDelayChannel"), adapterAccessor.getPropertyValue("outputChannel")); + Expression expression = TestUtils.getPropertyValue(adapter, "source.expression", Expression.class); + assertEquals("'fixedDelayTest'", expression.getExpressionString()); } @Test public void fixedRate() { - ScheduledMessageProducer producer = context.getBean("fixedRateProducer", ScheduledMessageProducer.class); - assertFalse(producer.isAutoStartup()); - DirectFieldAccessor producerAccessor = new DirectFieldAccessor(producer); - Trigger trigger = (Trigger) producerAccessor.getPropertyValue("trigger"); + SourcePollingChannelAdapter adapter = context.getBean("fixedRateProducer", SourcePollingChannelAdapter.class); + assertFalse(adapter.isAutoStartup()); + DirectFieldAccessor adapterAccessor = new DirectFieldAccessor(adapter); + Trigger trigger = TestUtils.getPropertyValue(adapter, "pollerMetadata.trigger", Trigger.class); assertEquals(PeriodicTrigger.class, trigger.getClass()); DirectFieldAccessor triggerAccessor = new DirectFieldAccessor(trigger); assertEquals(5678L, triggerAccessor.getPropertyValue("period")); assertEquals(Boolean.TRUE, triggerAccessor.getPropertyValue("fixedRate")); - assertEquals(context.getBean("fixedRateChannel"), producerAccessor.getPropertyValue("outputChannel")); - Expression payloadExpression = (Expression) new DirectFieldAccessor( - producerAccessor.getPropertyValue("task")).getPropertyValue("payloadExpression"); - assertEquals("'fixedRateTest'", payloadExpression.getExpressionString()); + assertEquals(context.getBean("fixedRateChannel"), adapterAccessor.getPropertyValue("outputChannel")); + Expression expression = TestUtils.getPropertyValue(adapter, "source.expression", Expression.class); + assertEquals("'fixedRateTest'", expression.getExpressionString()); } @Test public void cron() { - ScheduledMessageProducer producer = context.getBean("cronProducer", ScheduledMessageProducer.class); - assertFalse(producer.isAutoStartup()); - DirectFieldAccessor producerAccessor = new DirectFieldAccessor(producer); - Trigger trigger = (Trigger) producerAccessor.getPropertyValue("trigger"); + SourcePollingChannelAdapter adapter = context.getBean("cronProducer", SourcePollingChannelAdapter.class); + assertFalse(adapter.isAutoStartup()); + DirectFieldAccessor adapterAccessor = new DirectFieldAccessor(adapter); + Trigger trigger = TestUtils.getPropertyValue(adapter, "pollerMetadata.trigger", Trigger.class); assertEquals(CronTrigger.class, trigger.getClass()); assertEquals("7 6 5 4 3 ?", new DirectFieldAccessor(new DirectFieldAccessor( trigger).getPropertyValue("sequenceGenerator")).getPropertyValue("expression")); - assertEquals(context.getBean("cronChannel"), producerAccessor.getPropertyValue("outputChannel")); - Expression payloadExpression = (Expression) new DirectFieldAccessor( - producerAccessor.getPropertyValue("task")).getPropertyValue("payloadExpression"); - assertEquals("'cronTest'", payloadExpression.getExpressionString()); + assertEquals(context.getBean("cronChannel"), adapterAccessor.getPropertyValue("outputChannel")); + Expression expression = TestUtils.getPropertyValue(adapter, "source.expression", Expression.class); + assertEquals("'cronTest'", expression.getExpressionString()); } @Test public void triggerRef() { - ScheduledMessageProducer producer = context.getBean("triggerRefProducer", ScheduledMessageProducer.class); - assertTrue(producer.isAutoStartup()); - DirectFieldAccessor producerAccessor = new DirectFieldAccessor(producer); - Trigger trigger = (Trigger) producerAccessor.getPropertyValue("trigger"); + SourcePollingChannelAdapter adapter = context.getBean("triggerRefProducer", SourcePollingChannelAdapter.class); + assertTrue(adapter.isAutoStartup()); + DirectFieldAccessor adapterAccessor = new DirectFieldAccessor(adapter); + Trigger trigger = TestUtils.getPropertyValue(adapter, "pollerMetadata.trigger", Trigger.class); assertEquals(context.getBean("customTrigger"), trigger); - assertEquals(context.getBean("triggerRefChannel"), producerAccessor.getPropertyValue("outputChannel")); - Expression payloadExpression = (Expression) new DirectFieldAccessor( - producerAccessor.getPropertyValue("task")).getPropertyValue("payloadExpression"); - assertEquals("'triggerRefTest'", payloadExpression.getExpressionString()); + assertEquals(context.getBean("triggerRefChannel"), adapterAccessor.getPropertyValue("outputChannel")); + Expression expression = TestUtils.getPropertyValue(adapter, "source.expression", Expression.class); + assertEquals("'triggerRefTest'", expression.getExpressionString()); } @Test @SuppressWarnings("unchecked") public void headerExpressions() { - ScheduledMessageProducer producer = context.getBean("headerExpressionsProducer", ScheduledMessageProducer.class); - assertFalse(producer.isAutoStartup()); - DirectFieldAccessor producerAccessor = new DirectFieldAccessor(producer); - Map headerExpressions = (Map) producerAccessor.getPropertyValue("headerExpressions"); + SourcePollingChannelAdapter adapter = context.getBean("headerExpressionsProducer", SourcePollingChannelAdapter.class); + assertFalse(adapter.isAutoStartup()); + Map headerExpressions = TestUtils.getPropertyValue(adapter, "source.headerExpressions", Map.class); assertEquals(2, headerExpressions.size()); assertEquals("6 * 7", headerExpressions.get("foo").getExpressionString()); assertEquals("x", headerExpressions.get("bar").getExpressionString()); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/ScheduledMessageProducerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/ExpressionEvaluatingMessageSourceIntegrationTests.java similarity index 68% rename from spring-integration-core/src/test/java/org/springframework/integration/endpoint/ScheduledMessageProducerTests.java rename to spring-integration-core/src/test/java/org/springframework/integration/endpoint/ExpressionEvaluatingMessageSourceIntegrationTests.java index 0113a00fb9..23a688d5fe 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/ScheduledMessageProducerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/ExpressionEvaluatingMessageSourceIntegrationTests.java @@ -25,21 +25,22 @@ import java.util.Map; import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; - import org.springframework.expression.Expression; import org.springframework.expression.common.LiteralExpression; import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.integration.Message; import org.springframework.integration.channel.QueueChannel; -import org.springframework.scheduling.Trigger; +import org.springframework.integration.config.ExpressionFactoryBean; +import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.scheduling.support.PeriodicTrigger; +import org.springframework.util.ErrorHandler; /** * @author Mark Fisher * @since 2.0 */ -public class ScheduledMessageProducerTests { +public class ExpressionEvaluatingMessageSourceIntegrationTests { private static final AtomicInteger counter = new AtomicInteger(); @@ -47,18 +48,31 @@ public class ScheduledMessageProducerTests { @Test public void test() throws Exception { QueueChannel channel = new QueueChannel(); - Trigger trigger = new PeriodicTrigger(100); - String payloadExpression = "'test-' + T(org.springframework.integration.endpoint.ScheduledMessageProducerTests).next()"; + String payloadExpression = "'test-' + T(org.springframework.integration.endpoint.ExpressionEvaluatingMessageSourceIntegrationTests).next()"; ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); scheduler.afterPropertiesSet(); Map headerExpressions = new HashMap(); headerExpressions.put("foo", new LiteralExpression("x")); headerExpressions.put("bar", new SpelExpressionParser().parseExpression("7 * 6")); - ScheduledMessageProducer producer = new ScheduledMessageProducer(trigger, payloadExpression); - producer.setHeaderExpressions(headerExpressions); - producer.setTaskScheduler(scheduler); - producer.setOutputChannel(channel); - producer.start(); + ExpressionFactoryBean factoryBean = new ExpressionFactoryBean(payloadExpression); + factoryBean.afterPropertiesSet(); + Expression expression = factoryBean.getObject(); + ExpressionEvaluatingMessageSource source = new ExpressionEvaluatingMessageSource(expression, Object.class); + source.setHeaderExpressions(headerExpressions); + SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter(); + adapter.setSource(source); + adapter.setTaskScheduler(scheduler); + PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setMaxMessagesPerPoll(3); + pollerMetadata.setTrigger(new PeriodicTrigger(60000)); + adapter.setPollerMetadata(pollerMetadata); + adapter.setOutputChannel(channel); + adapter.setErrorHandler(new ErrorHandler() { + public void handleError(Throwable t) { + throw new IllegalStateException("unexpected exception in test", t); + } + }); + adapter.start(); List> messages = new ArrayList>(); for (int i = 0; i < 3; i++) { messages.add(channel.receive(1000)); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/ExpressionEvaluatingMessageSourceTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/ExpressionEvaluatingMessageSourceTests.java new file mode 100644 index 0000000000..00fd458e3a --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/ExpressionEvaluatingMessageSourceTests.java @@ -0,0 +1,57 @@ +/* + * Copyright 2002-2010 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.endpoint; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +import org.junit.Test; +import org.springframework.core.convert.ConversionFailedException; +import org.springframework.expression.Expression; +import org.springframework.expression.ExpressionParser; +import org.springframework.expression.common.LiteralExpression; +import org.springframework.expression.spel.standard.SpelExpressionParser; +import org.springframework.integration.Message; + +/** + * @author Mark Fisher + * @since 2.0 + */ +public class ExpressionEvaluatingMessageSourceTests { + + private static final ExpressionParser parser = new SpelExpressionParser(); + + + @Test + public void literalExpression() { + Expression expression = new LiteralExpression("foo"); + ExpressionEvaluatingMessageSource source = + new ExpressionEvaluatingMessageSource(expression, String.class); + Message message = source.receive(); + assertNotNull(message); + assertEquals("foo", message.getPayload()); + } + + @Test(expected = ConversionFailedException.class) + public void unexpectedType() { + Expression expression = new LiteralExpression("foo"); + ExpressionEvaluatingMessageSource source = + new ExpressionEvaluatingMessageSource(expression, Integer.class); + source.receive(); + } + +}