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`.

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

@@ -30,13 +30,13 @@ 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.context.IntegrationObjectSupport;
 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.jmx.export.annotation.ManagedResource;
 import org.springframework.messaging.Message;
@@ -81,7 +81,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;
 
@@ -325,10 +325,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));
 	}
 
@@ -337,8 +339,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);
 		}
@@ -350,6 +351,17 @@ 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);
@@ -363,10 +375,11 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
 	 * This behavior is dictated by the avoidance of invocation thread overload.
 	 */
 	@Override
-	public void reschedulePersistedMessages() {
+	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);
@@ -377,6 +390,7 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
 						releaseMessage(message);
 					}
 				}
+
 			}, new Date());
 		}
 	}

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
@@ -91,8 +91,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);
 	}
 
 	@Override

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

@@ -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;
@@ -37,6 +39,7 @@ import org.springframework.context.event.ContextRefreshedEvent;
 import org.springframework.context.support.StaticApplicationContext;
 import org.springframework.expression.Expression;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.integration.channel.QueueChannel;
 import org.springframework.messaging.MessageHandlingException;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.MessagePublishingErrorHandler;
@@ -437,6 +440,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<>("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);

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));
 	}
+
 }