INT-3560: Fix DelayHandler for Duplicate Messages

JIRA: https://jira.spring.io/browse/INT-3560

Previously, messages arriving at the delayer before the
the context was initialized would be emitted twice after
the context `refresh()` or a JMX invocation of `reschedulePersistedMessages()`.

When using a `SimpleMessageStore`, the "re" scheduled
message from the context refreshed event (or a JMX invocation)
would be re-handled unconditionally.

This was due to incorrect logic to handle the way the `SMS` stores messages.

Change the logic to correctly handle (ignore)  duplicate scheduled
releases when using `SMS`.

Conflicts:
	spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java
	spring-integration-core/src/test/java/org/springframework/integration/handler/DelayHandlerTests.java

spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java

@@ -30,7 +30,6 @@ import org.springframework.expression.EvaluationContext;
 import org.springframework.expression.EvaluationException;
 import org.springframework.expression.Expression;
 import org.springframework.expression.ExpressionParser;
-import org.springframework.expression.spel.SpelParserConfiguration;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.integration.Message;
 import org.springframework.integration.MessageHandlingException;
@@ -41,6 +40,7 @@ import org.springframework.integration.expression.ExpressionUtils;
 import org.springframework.integration.store.MessageGroup;
 import org.springframework.integration.store.MessageGroupStore;
 import org.springframework.integration.store.MessageStore;
+import org.springframework.integration.store.SimpleMessageGroup;
 import org.springframework.integration.store.SimpleMessageStore;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.jmx.export.annotation.ManagedResource;
@@ -83,7 +83,7 @@ import org.springframework.util.CollectionUtils;
 public class DelayHandler extends AbstractReplyProducingMessageHandler implements DelayHandlerManagement,
 		ApplicationListener<ContextRefreshedEvent> {
 
-	private static final ExpressionParser expressionParser = new SpelExpressionParser(new SpelParserConfiguration(true, true));
+	private static final ExpressionParser expressionParser = new SpelExpressionParser();
 
 	private final String messageGroupId;
 
@@ -310,9 +310,12 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
 		final Message<?> messageToSchedule = delayedMessage;
 
 		this.getTaskScheduler().schedule(new Runnable() {
+
+			@Override
 			public void run() {
 				releaseMessage(messageToSchedule);
 			}
+
 		}, new Date(messageWrapper.getRequestDate() + delay));
 	}
 
@@ -321,8 +324,7 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
 	}
 
 	private void doReleaseMessage(Message<?> message) {
-		if (this.messageStore instanceof SimpleMessageStore
-				|| ((MessageStore) this.messageStore).removeMessage(message.getHeaders().getId()) != null) {
+		if (removeDelayedMessageFromMessageStore(message)) {
 			this.messageStore.removeMessageFromGroup(this.messageGroupId, message);
 			this.handleMessageInternal(message);
 		}
@@ -334,6 +336,18 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
 		}
 	}
 
+	private boolean removeDelayedMessageFromMessageStore(Message<?> message) {
+		if (this.messageStore instanceof SimpleMessageStore) {
+			SimpleMessageGroup messageGroup =
+					(SimpleMessageGroup) this.messageStore.getMessageGroup(this.messageGroupId);
+			return messageGroup.remove(message);
+		}
+		else {
+			return ((MessageStore) this.messageStore).removeMessage(message.getHeaders().getId()) != null;
+		}
+	}
+
+	@Override
 	public int getDelayedMessageCount() {
 		return this.messageStore.messageGroupSize(this.messageGroupId);
 	}
@@ -345,10 +359,13 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
 	 * and schedules task about 'delay' logic.
 	 * This behavior is dictated by the avoidance of invocation thread overload.
 	 */
-	public void reschedulePersistedMessages() {
+	@Override
+	public synchronized void reschedulePersistedMessages() {
 		MessageGroup messageGroup = this.messageStore.getMessageGroup(this.messageGroupId);
 		for (final Message<?> message : messageGroup.getMessages()) {
 			this.getTaskScheduler().schedule(new Runnable() {
+
+				@Override
 				public void run() {
 					long delay = determineDelayForMessage(message);
 					if (delay > 0) {
@@ -358,6 +375,7 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
 						releaseMessage(message);
 					}
 				}
+
 			}, new Date());
 		}
 	}
@@ -374,6 +392,7 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
 	 *
 	 * @see #reschedulePersistedMessages
 	 */
+	@Override
 	public void onApplicationEvent(ContextRefreshedEvent event) {
 		if (!this.initialized.getAndSet(true)) {
 			this.reschedulePersistedMessages();
@@ -390,6 +409,7 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
 	 */
 	private class ReleaseMessageHandler implements MessageHandler {
 
+		@Override
 		public void handleMessage(Message<?> message) throws MessagingException {
 			DelayHandler.this.doReleaseMessage(message);
 		}

spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2012 the original author or authors.
+ * Copyright 2002-2014 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
@@ -87,8 +87,8 @@ public class SimpleMessageGroup implements MessageGroup {
 		addMessage(message);
 	}
 
-	public void remove(Message<?> message) {
-		messages.remove(message);
+	public boolean remove(Message<?> message) {
+		return this.messages.remove(message);
 	}
 
 	public int getLastReleasedMessageSequenceNumber() {
@@ -148,4 +148,4 @@ public class SimpleMessageGroup implements MessageGroup {
 				", lastModified=" + lastModified +
 				'}';
 	}
-}
+}

spring-integration-core/src/test/java/org/springframework/integration/handler/DelayHandlerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2013 the original author or authors.
+ * Copyright 2002-2014 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.
@@ -17,7 +17,9 @@
 package org.springframework.integration.handler;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNotSame;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.Mockito.mock;
@@ -42,6 +44,7 @@ import org.springframework.integration.MessageDeliveryException;
 import org.springframework.integration.MessageHandlingException;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.MessagePublishingErrorHandler;
+import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.context.IntegrationContextUtils;
 import org.springframework.integration.core.MessageHandler;
 import org.springframework.integration.message.GenericMessage;
@@ -231,6 +234,7 @@ public class DelayHandlerTests {
 
 		final CountDownLatch latch = new CountDownLatch(1);
 		new Thread(new Runnable() {
+			@Override
 			public void run() {
 				try {
 					taskScheduler.getScheduledExecutor().awaitTermination(10000, TimeUnit.MILLISECONDS);
@@ -255,6 +259,7 @@ public class DelayHandlerTests {
 
 		final CountDownLatch latch = new CountDownLatch(1);
 		new Thread(new Runnable() {
+			@Override
 			public void run() {
 				try {
 					taskScheduler.getScheduledExecutor().awaitTermination(10000, TimeUnit.MILLISECONDS);
@@ -274,6 +279,7 @@ public class DelayHandlerTests {
 		this.startDelayerHandler();
 		output.unsubscribe(resultHandler);
 		output.subscribe(new MessageHandler() {
+			@Override
 			public void handleMessage(Message<?> message) {
 				throw new UnsupportedOperationException("intentional test failure");
 			}
@@ -293,6 +299,7 @@ public class DelayHandlerTests {
 		output.unsubscribe(resultHandler);
 		errorChannel.subscribe(resultHandler);
 		output.subscribe(new MessageHandler() {
+			@Override
 			public void handleMessage(Message<?> message) {
 				throw new UnsupportedOperationException("intentional test failure");
 			}
@@ -326,6 +333,7 @@ public class DelayHandlerTests {
 		output.unsubscribe(resultHandler);
 		customErrorChannel.subscribe(resultHandler);
 		output.subscribe(new MessageHandler() {
+			@Override
 			public void handleMessage(Message<?> message) {
 				throw new UnsupportedOperationException("intentional test failure");
 			}
@@ -357,6 +365,7 @@ public class DelayHandlerTests {
 		output.unsubscribe(resultHandler);
 		defaultErrorChannel.subscribe(resultHandler);
 		output.subscribe(new MessageHandler() {
+			@Override
 			public void handleMessage(Message<?> message) {
 				throw new UnsupportedOperationException("intentional test failure");
 			}
@@ -418,6 +427,7 @@ public class DelayHandlerTests {
 	public void testDoubleOnApplicationEvent() throws Exception {
 		this.delayHandler = Mockito.spy(this.delayHandler);
 		Mockito.doAnswer(new Answer<Object>() {
+			@Override
 			public Object answer(InvocationOnMock invocation) throws Throwable {
 				return null;
 			}
@@ -437,6 +447,25 @@ public class DelayHandlerTests {
 		this.delayHandler.handleMessage(new GenericMessage<String>("test"));
 	}
 
+	@Test //INT-3560
+	/*
+	 It's difficult to test it from real ctx, because any async process from 'inbound-channel-adapter'
+	 can't achieve the DelayHandler before the main thread emits 'ContextRefreshedEvent'.
+	 */
+	public void testRescheduleAndHandleAtTheSameTime() throws Exception {
+		QueueChannel results = new QueueChannel();
+		delayHandler.setOutputChannel(results);
+		this.delayHandler.setDefaultDelay(100);
+		startDelayerHandler();
+
+		this.input.send(new GenericMessage<String>("foo"));
+		this.delayHandler.reschedulePersistedMessages();
+		Message<?> message = results.receive(10000);
+		assertNotNull(message);
+		message = results.receive(500);
+		assertNull(message);
+	}
+
 	private void waitForLatch(long timeout) {
 		try {
 			this.latch.await(timeout, TimeUnit.MILLISECONDS);
@@ -456,6 +485,7 @@ public class DelayHandlerTests {
 
 		private volatile Thread lastThread;
 
+		@Override
 		public void handleMessage(Message<?> message) {
 			this.lastMessage = message;
 			this.lastThread = Thread.currentThread();

spring-integration-groovy/src/test/java/org/springframework/integration/groovy/GroovyControlBusIntegrationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2012 the original author or authors.
+ * Copyright 2002-2014 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
@@ -44,9 +44,6 @@ public class GroovyControlBusIntegrationTests {
 	@Autowired
 	private MessageChannel controlBus;
 
-	@Autowired
-	private PollableChannel controlBusOutput;
-
 	@Autowired
 	private PollableChannel output;
 
@@ -72,7 +69,8 @@ public class GroovyControlBusIntegrationTests {
 		Message<?> message = MessageBuilder.withPayload(scriptSource.getScriptAsString()).build();
 		this.controlBus.send(message);
 
-		assertNotNull(this.output.receive(1000));
-		assertNotNull(this.output.receive(1000));
+		assertNotNull(this.output.receive(10000));
+		assertNotNull(this.output.receive(10000));
 	}
+
 }