diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingConsumerEndpointTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingConsumerEndpointTests.java index f7b569dff3..d1c2316cd1 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingConsumerEndpointTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingConsumerEndpointTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-2018 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. @@ -19,10 +19,6 @@ package org.springframework.integration.endpoint; import static org.junit.Assert.assertEquals; import static org.mockito.Mockito.mock; -import java.util.Date; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import org.junit.After; @@ -33,12 +29,11 @@ import org.mockito.Mockito; import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.MessageRejectedException; import org.springframework.integration.support.MessagingExceptionWrapper; +import org.springframework.integration.test.util.OnlyOnceTrigger; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.GenericMessage; -import org.springframework.scheduling.Trigger; -import org.springframework.scheduling.TriggerContext; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.util.ErrorHandler; @@ -46,131 +41,132 @@ import org.springframework.util.ErrorHandler; * @author Iwein Fuld * @author Mark Fisher * @author Kiel Boatman + * @author Artem Bilan */ -@SuppressWarnings("unchecked") +@SuppressWarnings({ "rawtypes", "unchecked" }) public class PollingConsumerEndpointTests { - private PollingConsumer endpoint; - - private final TestTrigger trigger = new TestTrigger(); + private final OnlyOnceTrigger trigger = new OnlyOnceTrigger(); private final TestConsumer consumer = new TestConsumer(); - @SuppressWarnings("rawtypes") - private final Message message = new GenericMessage("test"); + private final Message message = new GenericMessage<>("test"); - @SuppressWarnings("rawtypes") - private final Message badMessage = new GenericMessage("bad"); + private final Message badMessage = new GenericMessage<>("bad"); private final TestErrorHandler errorHandler = new TestErrorHandler(); - private PollableChannel channelMock; - private final ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); + private PollingConsumer endpoint; + + private PollableChannel channelMock; + @Before - public void init() throws Exception { - channelMock = Mockito.mock(PollableChannel.class); - consumer.counter.set(0); - trigger.reset(); - endpoint = new PollingConsumer(channelMock, consumer); - taskScheduler.setPoolSize(5); - endpoint.setErrorHandler(errorHandler); - endpoint.setTaskScheduler(taskScheduler); - endpoint.setTrigger(trigger); - endpoint.setBeanFactory(mock(BeanFactory.class)); - endpoint.setReceiveTimeout(-1); - endpoint.afterPropertiesSet(); - taskScheduler.afterPropertiesSet(); + public void init() { + this.channelMock = mock(PollableChannel.class); + this.endpoint = new PollingConsumer(this.channelMock, this.consumer); + this.taskScheduler.setPoolSize(5); + this.endpoint.setErrorHandler(this.errorHandler); + this.endpoint.setTaskScheduler(this.taskScheduler); + this.endpoint.setTrigger(this.trigger); + this.endpoint.setBeanFactory(mock(BeanFactory.class)); + this.endpoint.setReceiveTimeout(-1); + this.endpoint.afterPropertiesSet(); + this.taskScheduler.afterPropertiesSet(); } @After - public void stop() throws Exception { + public void stop() { taskScheduler.destroy(); } @Test public void singleMessage() { - Mockito.when(channelMock.receive()).thenReturn(message); - endpoint.setMaxMessagesPerPoll(1); - endpoint.start(); - trigger.await(); - endpoint.stop(); - assertEquals(1, consumer.counter.get()); + Mockito.when(this.channelMock.receive()).thenReturn(this.message); + this.endpoint.setMaxMessagesPerPoll(1); + this.endpoint.start(); + this.trigger.await(); + this.endpoint.stop(); + assertEquals(1, this.consumer.counter.get()); } @Test public void multipleMessages() { - Mockito.when(channelMock.receive()).thenReturn(message, message, message, message, message); - endpoint.setMaxMessagesPerPoll(5); - endpoint.start(); - trigger.await(); - endpoint.stop(); - assertEquals(5, consumer.counter.get()); + Mockito.when(this.channelMock.receive()) + .thenReturn(this.message, this.message, this.message, this.message, this.message); + this.endpoint.setMaxMessagesPerPoll(5); + this.endpoint.start(); + this.trigger.await(); + this.endpoint.stop(); + assertEquals(5, this.consumer.counter.get()); } @Test - public void multipleMessages_underrun() { - Mockito.when(channelMock.receive()).thenReturn(message, message, message, message, message, null); - endpoint.setMaxMessagesPerPoll(6); - endpoint.start(); - trigger.await(); - endpoint.stop(); - assertEquals(5, consumer.counter.get()); + public void multipleMessages_under_run() { + Mockito.when(this.channelMock.receive()) + .thenReturn(this.message, this.message, this.message, this.message, this.message, null); + this.endpoint.setMaxMessagesPerPoll(6); + this.endpoint.start(); + this.trigger.await(); + this.endpoint.stop(); + assertEquals(5, this.consumer.counter.get()); } @Test - public void heavierLoadTest() throws Exception { + public void heavierLoadTest() { for (int i = 0; i < 1000; i++) { - this.init(); - this.multipleMessages(); - this.stop(); + init(); + this.trigger.reset(); + this.consumer.counter.set(0); + multipleMessages(); + stop(); } } @Test(expected = MessageRejectedException.class) public void rejectedMessage() throws Throwable { - Mockito.when(channelMock.receive()).thenReturn(badMessage); - endpoint.start(); - trigger.await(); - endpoint.stop(); - assertEquals(1, consumer.counter.get()); - errorHandler.throwLastErrorIfAvailable(); + Mockito.when(this.channelMock.receive()).thenReturn(this.badMessage); + this.endpoint.start(); + this.trigger.await(); + this.endpoint.stop(); + assertEquals(1, this.consumer.counter.get()); + this.errorHandler.throwLastErrorIfAvailable(); } @Test(expected = MessageRejectedException.class) public void droppedMessage_onePerPoll() throws Throwable { - Mockito.when(channelMock.receive()).thenReturn(badMessage); - endpoint.setMaxMessagesPerPoll(10); - endpoint.start(); - trigger.await(); - endpoint.stop(); - assertEquals(1, consumer.counter.get()); - errorHandler.throwLastErrorIfAvailable(); + Mockito.when(this.channelMock.receive()).thenReturn(this.badMessage); + this.endpoint.setMaxMessagesPerPoll(10); + this.endpoint.start(); + this.trigger.await(); + this.endpoint.stop(); + assertEquals(1, this.consumer.counter.get()); + this.errorHandler.throwLastErrorIfAvailable(); } @Test public void blockingSourceTimedOut() { // we don't need to await the timeout, returning null suffices - Mockito.when(channelMock.receive()).thenReturn(null); - endpoint.setReceiveTimeout(1); - endpoint.start(); - trigger.await(); - endpoint.stop(); - assertEquals(0, consumer.counter.get()); + Mockito.when(this.channelMock.receive()).thenReturn(null); + this.endpoint.setReceiveTimeout(1); + this.endpoint.start(); + this.trigger.await(); + this.endpoint.stop(); + assertEquals(0, this.consumer.counter.get()); } @Test public void blockingSourceNotTimedOut() { - Mockito.when(channelMock.receive(Mockito.eq(1L))).thenReturn(message); - endpoint.setReceiveTimeout(1); - endpoint.setMaxMessagesPerPoll(1); - endpoint.start(); - trigger.await(); - endpoint.stop(); - assertEquals(1, consumer.counter.get()); + Mockito.when(this.channelMock.receive(Mockito.eq(1L))).thenReturn(this.message); + this.endpoint.setReceiveTimeout(1); + this.endpoint.setMaxMessagesPerPoll(1); + this.endpoint.start(); + this.trigger.await(); + this.endpoint.stop(); + assertEquals(1, this.consumer.counter.get()); } @@ -192,45 +188,6 @@ public class PollingConsumerEndpointTests { } - private static class TestTrigger implements Trigger { - - private final AtomicBoolean hasRun = new AtomicBoolean(); - - private volatile CountDownLatch latch = new CountDownLatch(1); - - - TestTrigger() { - super(); - } - - @Override - public Date nextExecutionTime(TriggerContext triggerContext) { - if (!this.hasRun.getAndSet(true)) { - return new Date(); - } - this.latch.countDown(); - return null; - } - - public void reset() { - this.latch = new CountDownLatch(1); - this.hasRun.set(false); - } - - public void await() { - try { - this.latch.await(5000, TimeUnit.MILLISECONDS); - if (latch.getCount() != 0) { - throw new RuntimeException("test latch.await() did not count down"); - } - } - catch (InterruptedException e) { - throw new RuntimeException("test latch.await() interrupted"); - } - } - } - - private static class TestErrorHandler implements ErrorHandler { private volatile Throwable lastError; @@ -252,6 +209,7 @@ public class PollingConsumerEndpointTests { this.lastError = null; throw t; } + } } diff --git a/spring-integration-test-support/src/main/java/org/springframework/integration/test/util/OnlyOnceTrigger.java b/spring-integration-test-support/src/main/java/org/springframework/integration/test/util/OnlyOnceTrigger.java index 8dd1b691b9..5cc086a5bd 100644 --- a/spring-integration-test-support/src/main/java/org/springframework/integration/test/util/OnlyOnceTrigger.java +++ b/spring-integration-test-support/src/main/java/org/springframework/integration/test/util/OnlyOnceTrigger.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2018 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. @@ -28,6 +28,8 @@ import org.springframework.scheduling.TriggerContext; * * @author Gunnar Hillert * @author Gary Russell + * @author Artem Bilan + * * @since 2.2 * */ @@ -47,7 +49,6 @@ public class OnlyOnceTrigger implements Trigger { @Override public Date nextExecutionTime(TriggerContext triggerContext) { - if (this.hasRun.getAndSet(true)) { this.latch.countDown(); return null; @@ -95,7 +96,7 @@ public class OnlyOnceTrigger implements Trigger { public void await() { try { - if (!this.latch.await(5000, TimeUnit.MILLISECONDS)) { + if (!this.latch.await(10000, TimeUnit.MILLISECONDS)) { throw new RuntimeException("test latch.await() did not count down"); } }