From c31a96d4cb8050c1c3614e834cc33a7fb2dbd32c Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 6 Oct 2016 17:13:29 -0400 Subject: [PATCH] INT-4132: Start MS Before Scheduling Polling Task JIRA: https://jira.spring.io/browse/INT-4132 The race condition is present when polling task may be ran before `MessageSource` has been started. * Swap the order of `start()` in the `SourcePollingChannelAdapter`. Since proxying is now applied only for the `MessageSource.receive()` it doesn't hurt to start it before actual proxying. Just because it is really should be started before performing its `receive()` * Prove the proper order with the mock test and protect ourselves for the future similar changes * Also swap the `stop()` order in the `SourcePollingChannelAdapter. We have to stop/cancel the polling task before discarding internal `MessageSource` lifecycle. For example with the current state we may close an underlying resource already, but still have the last polling tick. That may cause any unexpected behaviour **Cherry-pick to 4.3.x** Polishing --- .../endpoint/SourcePollingChannelAdapter.java | 4 +- ...PollingChannelAdapterFactoryBeanTests.java | 122 ++++++++++++------ 2 files changed, 82 insertions(+), 44 deletions(-) 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 6b2c42f4a8..cec3b059ad 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 @@ -152,19 +152,19 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint @Override protected void doStart() { - super.doStart(); if (this.source instanceof Lifecycle) { ((Lifecycle) this.source).start(); } + super.doStart(); } @Override protected void doStop() { + super.doStop(); if (this.source instanceof Lifecycle) { ((Lifecycle) this.source).stop(); } - super.doStop(); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBeanTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBeanTests.java index 7ad0b12f5e..9e0ad823b1 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBeanTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBeanTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2016 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,10 @@ package org.springframework.integration.config; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; +import static org.mockito.BDDMockito.willAnswer; +import static org.mockito.Matchers.any; import static org.mockito.Matchers.contains; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.spy; @@ -37,12 +40,10 @@ import org.aopalliance.intercept.MethodInterceptor; import org.aopalliance.intercept.MethodInvocation; import org.apache.commons.logging.Log; import org.junit.Test; -import org.mockito.Mockito; -import org.mockito.invocation.InvocationOnMock; -import org.mockito.stubbing.Answer; import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.BeanFactory; +import org.springframework.context.Lifecycle; import org.springframework.integration.channel.MessagePublishingErrorHandler; import org.springframework.integration.channel.NullChannel; import org.springframework.integration.channel.QueueChannel; @@ -54,6 +55,8 @@ import org.springframework.integration.test.util.TestUtils.TestApplicationContex import org.springframework.messaging.Message; import org.springframework.messaging.MessagingException; import org.springframework.messaging.support.GenericMessage; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.scheduling.Trigger; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.scheduling.support.PeriodicTrigger; import org.springframework.util.ClassUtils; @@ -72,15 +75,13 @@ public class SourcePollingChannelAdapterFactoryBeanTests { factoryBean.setBeanFactory(context.getBeanFactory()); factoryBean.setBeanClassLoader(ClassUtils.getDefaultClassLoader()); factoryBean.setOutputChannel(outputChannel); - factoryBean.setSource(new TestSource()); + factoryBean.setSource(() -> new GenericMessage<>("test")); PollerMetadata pollerMetadata = new PollerMetadata(); List adviceChain = new ArrayList(); final AtomicBoolean adviceApplied = new AtomicBoolean(false); - adviceChain.add(new MethodInterceptor() { - public Object invoke(MethodInvocation invocation) throws Throwable { - adviceApplied.set(true); - return invocation.proceed(); - } + adviceChain.add((MethodInterceptor) invocation -> { + adviceApplied.set(true); + return invocation.proceed(); }); pollerMetadata.setTrigger(new PeriodicTrigger(5000)); pollerMetadata.setMaxMessagesPerPoll(1); @@ -95,7 +96,6 @@ public class SourcePollingChannelAdapterFactoryBeanTests { assertTrue("adviceChain was not applied", adviceApplied.get()); } - @SuppressWarnings("rawtypes") @Test public void testTransactionalAdviceChain() throws Throwable { SourcePollingChannelAdapterFactoryBean factoryBean = new SourcePollingChannelAdapterFactoryBean(); @@ -104,31 +104,25 @@ public class SourcePollingChannelAdapterFactoryBeanTests { factoryBean.setBeanFactory(context.getBeanFactory()); factoryBean.setBeanClassLoader(ClassUtils.getDefaultClassLoader()); factoryBean.setOutputChannel(outputChannel); - factoryBean.setSource(new TestSource()); + factoryBean.setSource(() -> new GenericMessage<>("test")); PollerMetadata pollerMetadata = new PollerMetadata(); List adviceChain = new ArrayList(); final AtomicBoolean adviceApplied = new AtomicBoolean(false); - adviceChain.add(new MethodInterceptor() { - public Object invoke(MethodInvocation invocation) throws Throwable { - adviceApplied.set(true); - return invocation.proceed(); - } + adviceChain.add((MethodInterceptor) invocation -> { + adviceApplied.set(true); + return invocation.proceed(); }); pollerMetadata.setTrigger(new PeriodicTrigger(5000)); pollerMetadata.setMaxMessagesPerPoll(1); final AtomicInteger count = new AtomicInteger(); final MethodInterceptor txAdvice = mock(MethodInterceptor.class); - adviceChain.add(new MethodInterceptor() { - public Object invoke(MethodInvocation invocation) throws Throwable { - count.incrementAndGet(); - return invocation.proceed(); - } + adviceChain.add((MethodInterceptor) invocation -> { + count.incrementAndGet(); + return invocation.proceed(); }); - when(txAdvice.invoke(Mockito.any(MethodInvocation.class))).thenAnswer(new Answer() { - public Object answer(InvocationOnMock invocation) throws Throwable { - count.incrementAndGet(); - return ((MethodInvocation) invocation.getArguments()[0]).proceed(); - } + when(txAdvice.invoke(any(MethodInvocation.class))).thenAnswer(invocation -> { + count.incrementAndGet(); + return invocation.getArgumentAt(0, MethodInvocation.class).proceed(); }); pollerMetadata.setAdviceChain(adviceChain); @@ -147,19 +141,16 @@ public class SourcePollingChannelAdapterFactoryBeanTests { public void testInterrupted() throws Exception { final CountDownLatch startLatch = new CountDownLatch(1); - MessageSource ms = new MessageSource() { - @Override - public Message receive() { - startLatch.countDown(); - try { - Thread.sleep(10000); - } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new MessagingException("Interrupted awaiting stopLatch", e); - } - return null; + MessageSource ms = () -> { + startLatch.countDown(); + try { + Thread.sleep(10000); } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new MessagingException("Interrupted awaiting stopLatch", e); + } + return null; }; SourcePollingChannelAdapter pollingChannelAdapter = new SourcePollingChannelAdapter(); @@ -199,11 +190,58 @@ public class SourcePollingChannelAdapterFactoryBeanTests { verify(adapterLogger).debug(contains("Poll interrupted - during stop()?")); } - private static class TestSource implements MessageSource { + @Test + public void testStartSourceBeforeRunPollingTask() { + TaskScheduler taskScheduler = mock(TaskScheduler.class); - public Message receive() { - return new GenericMessage("test"); + willAnswer(invocation -> { + Runnable task = invocation.getArgumentAt(0, Runnable.class); + task.run(); + return null; + }) + .given(taskScheduler) + .schedule(any(Runnable.class), any(Trigger.class)); + + SourcePollingChannelAdapter pollingChannelAdapter = new SourcePollingChannelAdapter(); + pollingChannelAdapter.setTaskScheduler(taskScheduler); + pollingChannelAdapter.setSource(new LifecycleMessageSource()); + pollingChannelAdapter.setMaxMessagesPerPoll(1); + QueueChannel outputChannel = new QueueChannel(); + pollingChannelAdapter.setOutputChannel(outputChannel); + pollingChannelAdapter.setBeanFactory(mock(BeanFactory.class)); + pollingChannelAdapter.afterPropertiesSet(); + pollingChannelAdapter.start(); + + Message receive = outputChannel.receive(10_000); + assertNotNull(receive); + assertEquals(true, receive.getPayload()); + pollingChannelAdapter.stop(); + } + + private static class LifecycleMessageSource implements MessageSource, Lifecycle { + + private volatile boolean running; + + @Override + public void start() { + this.running = true; } + + @Override + public void stop() { + this.running = false; + } + + @Override + public boolean isRunning() { + return this.running; + } + + @Override + public Message receive() { + return new GenericMessage<>(isRunning()); + } + } }