From dcb52f3e3b99af21fc1df924a9abe234385fc9df Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 5 Dec 2014 19:06:04 +0200 Subject: [PATCH] INT-3571: Propagate Lifecycle from SPCA to MS JIRA: https://jira.spring.io/browse/INT-3571 **Cherry-pick to 4.0.x** Introduce `LifecycleMessageSource` for `maint` versions to avoid undesired side-effects with already existing `... implements MessageSource, Lifecycle`. --- .../core/LifecycleMessageSource.java | 29 +++++++ .../endpoint/SourcePollingChannelAdapter.java | 20 +++++ .../endpoint/PollingLifecycleTests.java | 75 ++++++++++++++++--- 3 files changed, 114 insertions(+), 10 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/core/LifecycleMessageSource.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/core/LifecycleMessageSource.java b/spring-integration-core/src/main/java/org/springframework/integration/core/LifecycleMessageSource.java new file mode 100644 index 0000000000..57dc6391f5 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/core/LifecycleMessageSource.java @@ -0,0 +1,29 @@ +/* + * Copyright 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 + * + * 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. + */ + +package org.springframework.integration.core; + +import org.springframework.context.Lifecycle; + +/** + * The {@link Lifecycle} marker interface for backward compatibility. + * Will be deprecated in 4.2 and removed in the future releases. + * + * @author Artem Bilan + * @since 4.0.6 + */ +public interface LifecycleMessageSource extends MessageSource, Lifecycle { +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java index 1f665191bd..3bd75f9e66 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java @@ -16,6 +16,8 @@ package org.springframework.integration.endpoint; +import org.springframework.context.Lifecycle; +import org.springframework.integration.core.LifecycleMessageSource; import org.springframework.integration.core.MessageSource; import org.springframework.integration.core.MessagingTemplate; import org.springframework.integration.history.MessageHistory; @@ -91,6 +93,24 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint ((NamedComponent) this.source).getComponentType() : "inbound-channel-adapter"; } + @Override + protected void doStart() { + if (this.source instanceof LifecycleMessageSource) { + ((Lifecycle) this.source).start(); + } + super.doStart(); + } + + + @Override + protected void doStop() { + if (this.source instanceof LifecycleMessageSource) { + ((Lifecycle) this.source).stop(); + } + super.doStop(); + } + + @Override protected void onInit() { Assert.notNull(this.source, "source must not be null"); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingLifecycleTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingLifecycleTests.java index 7e03bba1a7..6bd72ce4cd 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingLifecycleTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingLifecycleTests.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. @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.endpoint; import static org.junit.Assert.assertNotNull; @@ -25,31 +26,37 @@ import static org.mockito.Mockito.times; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import org.junit.Before; import org.junit.Test; import org.mockito.Mockito; + import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.config.ConfigurableBeanFactory; -import org.springframework.messaging.Message; -import org.springframework.messaging.MessagingException; +import org.springframework.integration.channel.NullChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean; import org.springframework.integration.config.TestErrorHandler; -import org.springframework.messaging.MessageHandler; +import org.springframework.integration.core.LifecycleMessageSource; import org.springframework.integration.core.MessageSource; -import org.springframework.messaging.support.GenericMessage; import org.springframework.integration.scheduling.PollerMetadata; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.MessagingException; +import org.springframework.messaging.support.GenericMessage; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.scheduling.support.PeriodicTrigger; /** * @author Oleg Zhurakousky * @author Gunnar Hillert - * + * @author Artem Bilan */ public class PollingLifecycleTests { + private ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); + private TestErrorHandler errorHandler = new TestErrorHandler(); @Before @@ -58,7 +65,7 @@ public class PollingLifecycleTests { } @Test - public void ensurePollerTaskStops() throws Exception{ + public void ensurePollerTaskStops() throws Exception { final CountDownLatch latch = new CountDownLatch(1); QueueChannel channel = new QueueChannel(); channel.send(new GenericMessage("foo")); @@ -89,7 +96,7 @@ public class PollingLifecycleTests { } @Test - public void ensurePollerTaskStopsForAdapter() throws Exception{ + public void ensurePollerTaskStopsForAdapter() throws Exception { final CountDownLatch latch = new CountDownLatch(1); QueueChannel channel = new QueueChannel(); @@ -118,7 +125,7 @@ public class PollingLifecycleTests { } @Test - public void ensurePollerTaskStopsForAdapterWithInterruptible() throws Exception{ + public void ensurePollerTaskStopsForAdapterWithInterruptible() throws Exception { final CountDownLatch latch = new CountDownLatch(2); QueueChannel channel = new QueueChannel(); @@ -136,7 +143,8 @@ public class PollingLifecycleTests { Thread.sleep(1000); latch.countDown(); } - } catch (InterruptedException e) { + } + catch (InterruptedException e) { coughtInterrupted.run(); } @@ -156,4 +164,51 @@ public class PollingLifecycleTests { Thread.sleep(1000); Mockito.verify(coughtInterrupted, times(1)).run(); } + + @Test + public void testAdapterLifecycleIsPropagatedToMessageSource() throws Exception { + SourcePollingChannelAdapterFactoryBean adapterFactory = new SourcePollingChannelAdapterFactoryBean(); + adapterFactory.setOutputChannel(new NullChannel()); + adapterFactory.setBeanFactory(mock(ConfigurableBeanFactory.class)); + PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setTrigger(new PeriodicTrigger(2000)); + adapterFactory.setPollerMetadata(pollerMetadata); + + final AtomicBoolean startInvoked = new AtomicBoolean(); + + final AtomicBoolean stopInvoked = new AtomicBoolean(); + + adapterFactory.setSource(new LifecycleMessageSource() { + + @Override + public void start() { + startInvoked.set(true); + } + + @Override + public void stop() { + stopInvoked.set(true); + } + + @Override + public boolean isRunning() { + return false; + } + + @Override + public Message receive() { + return null; + } + + }); + + SourcePollingChannelAdapter adapter = adapterFactory.getObject(); + adapter.setTaskScheduler(this.taskScheduler); + adapter.start(); + adapter.stop(); + + assertTrue(startInvoked.get()); + assertTrue(stopInvoked.get()); + } + }