From 8c2b9184a5b072189a51f621e1a2ee046e2c6409 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Tue, 20 Jul 2010 19:37:12 +0000 Subject: [PATCH] INT-650 initial commit of TriggeredMessagePublisher and a test case --- .../endpoint/TriggeredMessagePublisher.java | 131 ++++++++++++++++++ .../TriggeredMessagePublisherTests.java | 83 +++++++++++ 2 files changed, 214 insertions(+) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/endpoint/TriggeredMessagePublisher.java create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/endpoint/TriggeredMessagePublisherTests.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/TriggeredMessagePublisher.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/TriggeredMessagePublisher.java new file mode 100644 index 0000000000..8c312d4b40 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/TriggeredMessagePublisher.java @@ -0,0 +1,131 @@ +/* + * 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.AccessException; +import org.springframework.expression.BeanResolver; +import org.springframework.expression.EvaluationContext; +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.message.MessageBuilder; +import org.springframework.scheduling.Trigger; +import org.springframework.util.Assert; +import org.springframework.util.CollectionUtils; + +/** + * @author Mark Fisher + * @since 2.0 + */ +public class TriggeredMessagePublisher extends MessageProducerSupport { + + private static final ExpressionParser PARSER = new SpelExpressionParser(); + + + private final Trigger trigger; + + private final MessagePublishingTask task; + + private volatile ScheduledFuture future; + + private volatile Map headerExpressionMap; + + private final StandardEvaluationContext context = new StandardEvaluationContext(); + + + public TriggeredMessagePublisher(Trigger trigger, String payloadExpression) { + Assert.notNull(trigger, "trigger must not be null"); + Assert.hasText(payloadExpression, "payloadExpression is required"); + this.trigger = trigger; + this.task = new MessagePublishingTask(PARSER.parseExpression(payloadExpression)); + } + + + public void setHeaderExpressions(Map headerExpressions) { + if (headerExpressions != null) { + Map parsedExpressions = new HashMap(); + for (Map.Entry entry : headerExpressions.entrySet()) { + parsedExpressions.put(entry.getKey(), PARSER.parseExpression(entry.getValue())); + } + this.headerExpressionMap = parsedExpressions; + } + } + + private Map evaluateHeaders() { + Map headers = new HashMap(); + if (this.headerExpressionMap != null) { + for (Map.Entry entry : this.headerExpressionMap.entrySet()) { + headers.put(entry.getKey(), entry.getValue().getValue(context)); + } + } + return headers; + } + + @Override + protected void onInit() { + super.onInit(); + final BeanFactory beanFactory = this.getBeanFactory(); + this.context.setBeanResolver(new BeanResolver() { + public Object resolve(EvaluationContext context, String beanName) throws AccessException { + return beanFactory.getBean(beanName); + } + }); + } + + @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 MessagePublishingTask implements Runnable { + + private final Expression payloadExpression; + + + private MessagePublishingTask(Expression payloadExpression) { + 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/test/java/org/springframework/integration/endpoint/TriggeredMessagePublisherTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/TriggeredMessagePublisherTests.java new file mode 100644 index 0000000000..8a2ffaee02 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/TriggeredMessagePublisherTests.java @@ -0,0 +1,83 @@ +/* + * 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 java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.Test; + +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.core.Message; +import org.springframework.scheduling.Trigger; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; +import org.springframework.scheduling.support.PeriodicTrigger; + +/** + * @author Mark Fisher + * @since 2.0 + */ +public class TriggeredMessagePublisherTests { + + private static final AtomicInteger counter = new AtomicInteger(); + + + @Test + public void test() throws Exception { + QueueChannel channel = new QueueChannel(); + Trigger trigger = new PeriodicTrigger(100); + String payloadExpression = "'test-' + T(org.springframework.integration.endpoint.TriggeredMessagePublisherTests).next()"; + ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); + scheduler.afterPropertiesSet(); + Map headerExpressions = new HashMap(); + headerExpressions.put("foo", "'x'"); + headerExpressions.put("bar", "7 * 6"); + TriggeredMessagePublisher publisher = new TriggeredMessagePublisher(trigger, payloadExpression); + publisher.setHeaderExpressions(headerExpressions); + publisher.setTaskScheduler(scheduler); + publisher.setOutputChannel(channel); + publisher.start(); + List> messages = new ArrayList>(); + for (int i = 0; i < 3; i++) { + messages.add(channel.receive(1000)); + } + scheduler.destroy(); + Message message1 = messages.get(0); + assertEquals("test-1", message1.getPayload()); + assertEquals("x", message1.getHeaders().get("foo")); + assertEquals(42, message1.getHeaders().get("bar")); + Message message2 = messages.get(1); + assertEquals("test-2", message2.getPayload()); + assertEquals("x", message2.getHeaders().get("foo")); + assertEquals(42, message2.getHeaders().get("bar")); + Message message3 = messages.get(2); + assertEquals("test-3", message3.getPayload()); + assertEquals("x", message3.getHeaders().get("foo")); + assertEquals(42, message3.getHeaders().get("bar")); + } + + + public static int next() { + return counter.incrementAndGet(); + } + +}