From 1c0fa63204d286308be81cbe71760e675fe522b5 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`. 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 --- .../integration/handler/DelayHandler.java | 30 ++++++++++++++--- .../integration/store/SimpleMessageGroup.java | 8 ++--- .../handler/DelayHandlerTests.java | 32 ++++++++++++++++++- .../GroovyControlBusIntegrationTests.java | 10 +++--- 4 files changed, 64 insertions(+), 16 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 0b2ed9a966..823f21812e 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,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 { - 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); } 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 bb266a1ea3..c8eeef8184 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 @@ -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 + '}'; } -} \ No newline at end of file +} 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 185550cde5..4f2594effb 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 @@ -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() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { return null; } @@ -437,6 +447,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); @@ -456,6 +485,7 @@ public class DelayHandlerTests { private volatile Thread lastThread; + @Override public void handleMessage(Message message) { this.lastMessage = message; this.lastThread = Thread.currentThread(); 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 4363f349a4..db3b93e36b 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)); } + }