INT-650 initial commit of TriggeredMessagePublisher and a test case

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<String, Expression> 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<String, String> headerExpressions) {
+		if (headerExpressions != null) {
+			Map<String, Expression> parsedExpressions = new HashMap<String, Expression>();
+			for (Map.Entry<String, String> entry : headerExpressions.entrySet()) {
+				parsedExpressions.put(entry.getKey(), PARSER.parseExpression(entry.getValue()));
+			}
+			this.headerExpressionMap = parsedExpressions;
+		}
+	}
+
+	private Map<String, Object> evaluateHeaders() {
+		Map<String, Object> headers = new HashMap<String, Object>();
+		if (this.headerExpressionMap != null) {
+			for (Map.Entry<String, Expression> 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<String, Object> headers = evaluateHeaders();
+				MessageBuilder<?> builder = MessageBuilder.withPayload(payload);
+				if (!CollectionUtils.isEmpty(headers)) {
+					builder.copyHeaders(headers);
+				}
+				sendMessage(builder.build());
+			}
+		}
+	}
+
+}

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<String, String> headerExpressions = new HashMap<String, String>();
+		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<Message<?>> messages = new ArrayList<Message<?>>();
+		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();
+	}
+
+}