From 2a419ba6d890083f121d1c9b519422590120e997 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 11 Jan 2013 10:27:11 +0200 Subject: [PATCH] INT-2878: MessageGroupStoreReaper's Lifecycle Fix `MessageGroupStoreReaper` continues to expire `MessageGroups` via `scheduled-task` when not `running`. * Add check `isRunning()` to `run()` & `destroy()` * Additional `MessageGroupStoreReaper` polishing * Add test-case JIRA: https://jira.springsource.org/browse/INT-2878 INT-2878: Polishing : PR Comments --- .../store/MessageGroupStoreReaper.java | 74 ++++++++++--------- .../store/MessageStoreReaperTests.java | 61 +++++++++------ 2 files changed, 78 insertions(+), 57 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStoreReaper.java b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStoreReaper.java index fe111c1b98..7f7392eddc 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStoreReaper.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStoreReaper.java @@ -1,11 +1,11 @@ /* - * Copyright 2002-2010 the original author or authors. - * + * Copyright 2002-2013 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. @@ -27,10 +27,10 @@ import org.springframework.util.Assert; * Convenient configurable component to allow explicit timed expiry of {@link MessageGroup} instances in a * {@link MessageGroupStore}. This component provides a no-args {@link #run()} method that is useful for remote or timed * execution and a {@link #destroy()} method that can optionally be called on shutdown. - * + * * @author Dave Syer * @author Dave Turanski - * + * @author Artem Bilan */ public class MessageGroupStoreReaper implements Runnable, DisposableBean, InitializingBean, SmartLifecycle { @@ -58,9 +58,9 @@ public class MessageGroupStoreReaper implements Runnable, DisposableBean, Initia } /** - * Flag to indicate that the stores should be expired when this component is destroyed (i.e. usuually when its + * Flag to indicate that the stores should be expired when this component is destroyed (i.e. usually when its * enclosing {@link ApplicationContext} is closed). - * + * * @param expireOnDestroy the flag value to set */ public void setExpireOnDestroy(boolean expireOnDestroy) { @@ -70,7 +70,7 @@ public class MessageGroupStoreReaper implements Runnable, DisposableBean, Initia /** * Timeout in milliseconds (default -1). If negative then no groups ever time out. If greater than zero then all * groups older than that value are expired when this component is {@link #run()}. - * + * * @param timeout the timeout to set */ public void setTimeout(long timeout) { @@ -78,8 +78,8 @@ public class MessageGroupStoreReaper implements Runnable, DisposableBean, Initia } /** - * A message group store to expire according the the other configurations. - * + * A message group store to expire according the other configurations. + * * @param messageGroupStore the {@link MessageGroupStore} to set */ public void setMessageGroupStore(MessageGroupStore messageGroupStore) { @@ -87,13 +87,18 @@ public class MessageGroupStoreReaper implements Runnable, DisposableBean, Initia } public void afterPropertiesSet() throws Exception { - Assert.state(messageGroupStore != null, "A MessageGroupStore must be provided"); + Assert.state(this.messageGroupStore != null, "A MessageGroupStore must be provided"); } public void destroy() throws Exception { - if (expireOnDestroy) { - logger.info("Expiring all messages from message group store: " + messageGroupStore); - messageGroupStore.expireMessageGroups(0); + if (this.expireOnDestroy) { + if (this.isRunning()) { + logger.info("Expiring all messages from message group store: " + this.messageGroupStore); + this.messageGroupStore.expireMessageGroups(0); + } + else { + logger.debug("'expireOnDestroy' is set to 'true' but the reaper is not currently running"); + } } } @@ -102,12 +107,12 @@ public class MessageGroupStoreReaper implements Runnable, DisposableBean, Initia * be executed by a scheduled task. */ public void run() { - if (timeout >= 0) { + if (this.timeout >= 0 && this.isRunning()) { if (logger.isDebugEnabled()) { - logger.debug("Expiring all messages older than timeout=" + timeout + " from message group store: " - + messageGroupStore); + logger.debug("Expiring all messages older than timeout=" + this.timeout + " from message group store: " + + this.messageGroupStore); } - messageGroupStore.expireMessageGroups(timeout); + this.messageGroupStore.expireMessageGroups(this.timeout); } } @@ -129,15 +134,17 @@ public class MessageGroupStoreReaper implements Runnable, DisposableBean, Initia public void stop() { this.lifecycleLock.lock(); try { - destroy(); - if (logger.isInfoEnabled()){ - logger.info("stopped " + this); + if (this.running) { + this.destroy(); + if (logger.isInfoEnabled()) { + logger.info("stopped " + this); + } } - running = false; - } + this.running = false; + } catch (Exception e) { - logger.error("failed to stop bean",e); - } + logger.error("failed to stop bean", e); + } finally { this.lifecycleLock.unlock(); } @@ -147,26 +154,26 @@ public class MessageGroupStoreReaper implements Runnable, DisposableBean, Initia this.lifecycleLock.lock(); try { return this.running; - } finally { + } + finally { this.lifecycleLock.unlock(); } } public int getPhase() { - return phase; + return this.phase; } public void setPhase(int phase) { this.phase = phase; } - public boolean isAutoStartup() { - return autoStartup; + return this.autoStartup; } - - public void setAutoStartup(boolean autostartup) { - this.autoStartup = autostartup; + + public void setAutoStartup(boolean autoStartup) { + this.autoStartup = autoStartup; } public void stop(Runnable callback) { @@ -179,4 +186,5 @@ public class MessageGroupStoreReaper implements Runnable, DisposableBean, Initia this.lifecycleLock.unlock(); } } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreReaperTests.java b/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreReaperTests.java index e6f9cf972d..115c037961 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreReaperTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreReaperTests.java @@ -1,11 +1,11 @@ /* - * Copyright 2002-2010 the original author or authors. - * + * Copyright 2002-2013 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. @@ -30,28 +30,29 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; /** * @author Dave Syer * @author Dave Turanski + * @author Artem Bilan */ @ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class) public class MessageStoreReaperTests { - + @Autowired @Qualifier("messageStore") private MessageGroupStore messageStore; - + @Autowired @Qualifier("messageStore2") private MessageGroupStore messageStore2; - + @Autowired @Qualifier("reaper2") private MessageGroupStoreReaper reaper2; - - @Autowired + + @Autowired @Qualifier("expiryCallback") private ExpiryCallback expiryCallback; - - @Autowired + + @Autowired @Qualifier("expiryCallback2") private ExpiryCallback expiryCallback2; @@ -64,33 +65,45 @@ public class MessageStoreReaperTests { assertEquals(0, messageStore.getMessageGroup("FOO").size()); assertEquals(1, expiryCallback.groups.size()); } - - @Test + + @Test public void testSmartLifecycle() throws Exception{ - - messageStore2.addMessageToGroup("FOO", new GenericMessage("foo")); + GenericMessage testMessage = new GenericMessage("foo"); + + messageStore2.addMessageToGroup("FOO", testMessage); assertEquals(1, messageStore2.getMessageGroup("FOO").size()); - + reaper2.setExpireOnDestroy(true); reaper2.setTimeout(0); - - + + if (!reaper2.isAutoStartup()){ reaper2.start(); } - + assertTrue(reaper2.isRunning()); - + //reaper timeout is set to 0, but need to ensure positive elapsed time Thread.sleep(1L); - + reaper2.stop(); assertTrue(!reaper2.isRunning()); - + assertEquals(0, messageStore2.getMessageGroup("FOO").size()); assertEquals(1, expiryCallback2.groups.size()); + + messageStore2.addMessageToGroup("FOO", testMessage); + assertEquals(1, messageStore2.getMessageGroup("FOO").size()); + reaper2.run(); + assertEquals(1, messageStore2.getMessageGroup("FOO").size()); + reaper2.stop(); + assertEquals(1, messageStore2.getMessageGroup("FOO").size()); + reaper2.start(); + reaper2.run(); + assertEquals(0, messageStore2.getMessageGroup("FOO").size()); + assertEquals(2, expiryCallback2.groups.size()); } - + public static class ExpiryCallback implements MessageGroupCallback { public final List groups = new ArrayList(); @@ -99,7 +112,7 @@ public class MessageStoreReaperTests { groups.add(group); messageGroupStore.removeMessageGroup(group.getGroupId()); } - + } }