From 3d43ad1aa973c1edad2e840403fe57a39c998bb6 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 14 Nov 2014 17:50:59 +0200 Subject: [PATCH] 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`. --- .../integration/handler/DelayHandler.java | 24 +++++++++++++++---- .../integration/store/SimpleMessageGroup.java | 6 ++--- .../handler/DelayHandlerTests.java | 22 +++++++++++++++++ .../GroovyControlBusIntegrationTests.java | 10 ++++---- 4 files changed, 48 insertions(+), 14 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java index d58ef501e0..21c6ce0b86 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java +++ b/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 { - 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()); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java index 57f47f7b12..fd1e8527b1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java +++ b/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 diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/DelayHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/DelayHandlerTests.java index 62554c8eab..2961116497 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/DelayHandlerTests.java +++ b/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("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); diff --git a/spring-integration-groovy/src/test/java/org/springframework/integration/groovy/GroovyControlBusIntegrationTests.java b/spring-integration-groovy/src/test/java/org/springframework/integration/groovy/GroovyControlBusIntegrationTests.java index abb8bc3fd0..064ba9aae7 100644 --- a/spring-integration-groovy/src/test/java/org/springframework/integration/groovy/GroovyControlBusIntegrationTests.java +++ b/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)); } + }