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

INT-1494 first step: scheduled-producer now creates an instance of SourcePollingChannelAdapter

Browse Source
This commit is contained in:
Mark Fisher
2010-10-17 10:35:30 -04:00
parent 4db40c79b1
commit bf9c8a15f0
10 changed files with 294 additions and 227 deletions
Download Patch File Download Diff File Expand all files Collapse all files

63
spring-integration-core/src/main/java/org/springframework/integration/config/xml/ScheduledProducerParser.java
View File

@@ -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<Element> headerElements = DomUtils.getChildElementsByTagName(element, "header");
if (!CollectionUtils.isEmpty(headerElements)) {
ManagedMap<String, Object> headerExpressions = new ManagedMap<String, Object>();

101
spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractMessageSource.java Normal file
View File

@@ -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<T> extends AbstractExpressionEvaluator implements MessageSource<T> {
private volatile Map<String, Expression> headerExpressions = Collections.emptyMap();
public void setHeaderExpressions(Map<String, Expression> headerExpressions) {
this.headerExpressions = (headerExpressions != null)
? headerExpressions : Collections.<String, Expression>emptyMap();
}
@SuppressWarnings("unchecked")
public final Message<T> receive() {
Message<T> message = null;
Object result = this.doReceive();
if (result == null) {
return null;
}
Map<String, Object> headers = this.evaluateHeaders();
if (result instanceof Message<?>) {
try {
message = (Message<T>) 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<T> 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<T> builder = MessageBuilder.withPayload(payload);
if (!CollectionUtils.isEmpty(headers)) {
builder.copyHeaders(headers);
}
message = builder.build();
}
return message;
}
private Map<String, Object> evaluateHeaders() {
Map<String, Object> results = new HashMap<String, Object>();
for (Map.Entry<String, Expression> 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();
}

43
spring-integration-core/src/main/java/org/springframework/integration/endpoint/ExpressionEvaluatingMessageSource.java Normal file
View File

@@ -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<T> extends AbstractMessageSource<T> {
private final Expression expression;
private final Class<T> expectedType;
public ExpressionEvaluatingMessageSource(Expression expression, Class<T> 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);
}
}

124
spring-integration-core/src/main/java/org/springframework/integration/endpoint/ScheduledMessageProducer.java
View File

@@ -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<String, Expression> headerExpressions = new HashMap<String, Expression>();
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<String, Expression> headerExpressions) {
synchronized (this.headerExpressions) {
this.headerExpressions.clear();
if (headerExpressions != null) {
this.headerExpressions.putAll(headerExpressions);
}
}
}
private Map<String, Object> evaluateHeaders() {
Map<String, Object> headers = new HashMap<String, Object>();
for (Map.Entry<String, Expression> 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<String, Expression> headerExpressions) {
this.payloadExpression = payloadExpression;
}
public void run() {
Object payload = this.payloadExpression.getValue(context);
if (payload != null) {
Map<String, Object> headers = evaluateHeaders();
MessageBuilder<?> builder = MessageBuilder.withPayload(payload);
if (!CollectionUtils.isEmpty(headers)) {
builder.copyHeaders(headers);
}
sendMessage(builder.build());
}
}
}
}

8
spring-integration-core/src/main/java/org/springframework/integration/util/AbstractExpressionEvaluator.java
View File

@@ -97,6 +97,14 @@ public abstract class AbstractExpressionEvaluator implements BeanFactoryAware {
return this.evaluateExpression(expression, input, (Class<?>) null);
}
protected <T> T evaluateExpression(Expression expression, Class<T> expectedType) {
return expression.getValue(this.evaluationContext, expectedType);
}
protected Object evaluateExpression(Expression expression) {
return expression.getValue(this.evaluationContext);
}
protected <T> T evaluateExpression(Expression expression, Object input, Class<T> expectedType) {
return expression.getValue(this.evaluationContext, input, expectedType);
}

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

@@ -2424,6 +2424,7 @@ Name of the header whose value to use.
</xsd:annotation>
<xsd:complexType>
<xsd:sequence>
<xsd:element name="poller" type="innerPollerType" minOccurs="0" maxOccurs="1"/>
<xsd:element name="header" type="headerSubElementType" minOccurs="0" maxOccurs="unbounded" />
</xsd:sequence>
<xsd:attribute name="id" type="xsd:ID" />

19
spring-integration-core/src/test/java/org/springframework/integration/config/xml/ScheduledProducerParserTests-context.xml
View File

@@ -19,18 +19,27 @@
<queue/>
</channel>
<scheduled-producer id="fixedDelayProducer" fixed-delay="1234" payload-expression="'fixedDelayTest'" channel="fixedDelayChannel" auto-startup="false"/>
<scheduled-producer id="fixedDelayProducer" payload-expression="'fixedDelayTest'" channel="fixedDelayChannel" auto-startup="false">
<poller fixed-delay="1234"/>
</scheduled-producer>
<scheduled-producer id="fixedRateProducer" fixed-rate="5678" payload-expression="'fixedRateTest'" channel="fixedRateChannel" auto-startup="false"/>
<scheduled-producer id="fixedRateProducer" payload-expression="'fixedRateTest'" channel="fixedRateChannel" auto-startup="false">
<poller fixed-rate="5678"/>
</scheduled-producer>
<scheduled-producer id="cronProducer" cron="7 6 5 4 3 ?" payload-expression="'cronTest'" channel="cronChannel" auto-startup="false"/>
<scheduled-producer id="cronProducer" payload-expression="'cronTest'" channel="cronChannel" auto-startup="false">
<poller cron="7 6 5 4 3 ?"/>
</scheduled-producer>
<scheduled-producer id="headerExpressionsProducer" fixed-delay="99" payload-expression="'headerExpressionsTest'" channel="headerExpressionsChannel" auto-startup="false">
<scheduled-producer id="headerExpressionsProducer" payload-expression="'headerExpressionsTest'" channel="headerExpressionsChannel" auto-startup="false">
<poller fixed-delay="99"/>
<header name="foo" expression="6 * 7"/>
<header name="bar" value="x"/>
</scheduled-producer>
<scheduled-producer id="triggerRefProducer" trigger="customTrigger" payload-expression="'triggerRefTest'" channel="triggerRefChannel"/>
<scheduled-producer id="triggerRefProducer" payload-expression="'triggerRefTest'" channel="triggerRefChannel">
<poller trigger="customTrigger"/>
</scheduled-producer>
<beans:bean id="customTrigger" class="org.springframework.scheduling.support.PeriodicTrigger">
<beans:constructor-arg value="9999"/>

71
spring-integration-core/src/test/java/org/springframework/integration/config/xml/ScheduledProducerParserTests.java
View File

@@ -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<String, Expression> headerExpressions = (Map<String, Expression>) producerAccessor.getPropertyValue("headerExpressions");
SourcePollingChannelAdapter adapter = context.getBean("headerExpressionsProducer", SourcePollingChannelAdapter.class);
assertFalse(adapter.isAutoStartup());
Map<String, Expression> 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());

34
spring-integration-core/src/test/java/org/springframework/integration/endpoint/ScheduledMessageProducerTests.java → spring-integration-core/src/test/java/org/springframework/integration/endpoint/ExpressionEvaluatingMessageSourceIntegrationTests.java
View File

@@ -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<String, Expression> headerExpressions = new HashMap<String, Expression>();
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<Object> source = new ExpressionEvaluatingMessageSource<Object>(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<Message<?>> messages = new ArrayList<Message<?>>();
for (int i = 0; i < 3; i++) {
messages.add(channel.receive(1000));

57
spring-integration-core/src/test/java/org/springframework/integration/endpoint/ExpressionEvaluatingMessageSourceTests.java Normal file
View File

@@ -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<String> source =
new ExpressionEvaluatingMessageSource<String>(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<Integer> source =
new ExpressionEvaluatingMessageSource<Integer>(expression, Integer.class);
source.receive();
}
}