From 06979d7678143c85ce59607920b9c60dc7ba88cc Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 14 Oct 2013 17:56:25 +0300 Subject: [PATCH] INT-3171: Polishing Aggregator tests * Increase waiting timeouts * Get rid of `Thread.sleep` when it's dangerous JIRA: https://jira.springsource.org/browse/INT-3171 --- .../aggregator/AggregatorTests.java | 59 +++++++------- .../aggregator/ConcurrentAggregatorTests.java | 77 +++++++++++-------- .../CorrelatingMessageHandlerTests.java | 24 +++--- ...regatorWithCustomReleaseStrategyTests.java | 16 ++-- 4 files changed, 92 insertions(+), 84 deletions(-) diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java index 73d7f19cb1..87ca0abc80 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java @@ -1,11 +1,11 @@ /* - * Copyright 2002-2011 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. @@ -19,9 +19,6 @@ import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertThat; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; - import org.junit.Before; import org.junit.Ignore; import org.junit.Test; @@ -59,12 +56,12 @@ public class AggregatorTests { Message message1 = createMessage(3, "ABC", 3, 1, replyChannel, null); Message message2 = createMessage(5, "ABC", 3, 2, replyChannel, null); Message message3 = createMessage(7, "ABC", 3, 3, replyChannel, null); - CountDownLatch latch = new CountDownLatch(3); + this.aggregator.handleMessage(message1); this.aggregator.handleMessage(message2); this.aggregator.handleMessage(message3); - latch.await(1000, TimeUnit.MILLISECONDS); - Message reply = replyChannel.receive(2000); + + Message reply = replyChannel.receive(10000); assertNotNull(reply); assertEquals(reply.getPayload(), 105); } @@ -77,7 +74,7 @@ public class AggregatorTests { Message message = createMessage(3, "ABC", 2, 1, replyChannel, null); this.aggregator.handleMessage(message); this.store.expireMessageGroups(-10000); - Message reply = replyChannel.receive(100); + Message reply = replyChannel.receive(1000); assertNull("No message should have been sent normally", reply); Message discardedMessage = discardChannel.receive(1000); assertNotNull("A message should have been discarded", discardedMessage); @@ -93,7 +90,7 @@ public class AggregatorTests { this.aggregator.handleMessage(message1); this.aggregator.handleMessage(message2); this.store.expireMessageGroups(-10000); - Message reply = replyChannel.receive(0); + Message reply = replyChannel.receive(1000); assertNotNull("A reply message should have been received", reply); assertEquals(15, reply.getPayload()); } @@ -115,11 +112,11 @@ public class AggregatorTests { aggregator.handleMessage(message4); aggregator.handleMessage(message2); @SuppressWarnings("unchecked") - Message reply1 = (Message) replyChannel1.receive(500); + Message reply1 = (Message) replyChannel1.receive(1000); assertNotNull(reply1); assertThat(reply1.getPayload(), is(105)); @SuppressWarnings("unchecked") - Message reply2 = (Message) replyChannel2.receive(500); + Message reply2 = (Message) replyChannel2.receive(1000); assertNotNull(reply2); assertThat(reply2.getPayload(), is(2431)); } @@ -133,14 +130,14 @@ public class AggregatorTests { this.aggregator.setDiscardChannel(discardChannel); this.aggregator.handleMessage(createMessage(1, 1, 1, 1, replyChannel, null)); - assertEquals(1, replyChannel.receive(100).getPayload()); + assertEquals(1, replyChannel.receive(1000).getPayload()); this.aggregator.handleMessage(createMessage(3, 2, 1, 1, replyChannel, null)); - assertEquals(3, replyChannel.receive(100).getPayload()); + assertEquals(3, replyChannel.receive(1000).getPayload()); this.aggregator.handleMessage(createMessage(4, 3, 1, 1, replyChannel, null)); - assertEquals(4, replyChannel.receive(100).getPayload()); + assertEquals(4, replyChannel.receive(1000).getPayload()); // next message with same correllation ID is discarded this.aggregator.handleMessage(createMessage(2, 1, 1, 1, replyChannel, null)); - assertEquals(2, discardChannel.receive(100).getPayload()); + assertEquals(2, discardChannel.receive(1000).getPayload()); } @Test @@ -152,15 +149,15 @@ public class AggregatorTests { this.aggregator.setDiscardChannel(discardChannel); this.aggregator.handleMessage(createMessage(1, 1, 1, 1, replyChannel, null)); - assertEquals(1, replyChannel.receive(100).getPayload()); + assertEquals(1, replyChannel.receive(1000).getPayload()); this.aggregator.handleMessage(createMessage(2, 2, 1, 1, replyChannel, null)); - assertEquals(2, replyChannel.receive(100).getPayload()); + assertEquals(2, replyChannel.receive(1000).getPayload()); this.aggregator.handleMessage(createMessage(3, 3, 1, 1, replyChannel, null)); - assertEquals(3, replyChannel.receive(100).getPayload()); + assertEquals(3, replyChannel.receive(1000).getPayload()); this.aggregator.handleMessage(createMessage(4, 4, 1, 1, replyChannel, null)); - assertEquals(4, replyChannel.receive(100).getPayload()); + assertEquals(4, replyChannel.receive(1000).getPayload()); this.aggregator.handleMessage(createMessage(5, 1, 1, 1, replyChannel, null)); - assertEquals(5, replyChannel.receive(100).getPayload()); + assertEquals(5, replyChannel.receive(1000).getPayload()); assertNull(discardChannel.receive(0)); } @@ -177,15 +174,13 @@ public class AggregatorTests { Message message2 = createMessage(5, "ABC", 3, 2, replyChannel, null); Message message3 = createMessage(7, "ABC", 3, 3, replyChannel, null); Message message4 = createMessage(7, "ABC", 3, 3, replyChannel, null); - CountDownLatch latch = new CountDownLatch(4); + this.aggregator.handleMessage(message1); this.aggregator.handleMessage(message2); this.aggregator.handleMessage(message3); this.aggregator.handleMessage(message4); - latch.await(1000, TimeUnit.MILLISECONDS); - // small wait to make sure the fourth message is received - Thread.sleep(10); - Message reply = replyChannel.receive(0); + + Message reply = replyChannel.receive(10000); assertNotNull("A message should be aggregated", reply); assertThat(((Integer) reply.getPayload()), is(105)); } @@ -197,14 +192,14 @@ public class AggregatorTests { Message message2 = createMessage(5, "ABC", 3, 2, replyChannel, null); Message message3 = createMessage(7, "ABC", 3, 3, replyChannel, null); Message message4 = createMessage(7, "ABC", 3, 3, replyChannel, null); - CountDownLatch latch = new CountDownLatch(4); + this.aggregator.handleMessage(message1); this.aggregator.handleMessage(message3); // duplicated sequence number, either message3 or message4 should be rejected this.aggregator.handleMessage(message4); this.aggregator.handleMessage(message2); - latch.await(1000, TimeUnit.MILLISECONDS); - Message reply = replyChannel.receive(0); + + Message reply = replyChannel.receive(10000); assertNotNull("A message should be aggregated", reply); assertThat(((Integer) reply.getPayload()), is(105)); } @@ -219,7 +214,7 @@ public class AggregatorTests { this.aggregator.handleMessage(message1); this.aggregator.handleMessage(message2); this.aggregator.handleMessage(message3); - Message reply = replyChannel.receive(500); + Message reply = replyChannel.receive(1000); assertNull(reply); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java index bc85fb573c..4e28e136bc 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 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. @@ -16,12 +16,20 @@ package org.springframework.integration.aggregator; +import static org.hamcrest.CoreMatchers.is; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; + import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import org.junit.Before; import org.junit.Ignore; import org.junit.Test; + import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.core.task.TaskExecutor; import org.springframework.integration.Message; @@ -35,13 +43,6 @@ import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.store.SimpleMessageStore; import org.springframework.integration.support.MessageBuilder; -import static org.hamcrest.CoreMatchers.is; - -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertNull; -import static org.junit.Assert.assertThat; - /** * @author Mark Fisher * @author Marius Bogoevici @@ -76,7 +77,9 @@ public class ConcurrentAggregatorTests { message2, latch)); this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message3, latch)); - latch.await(10000, TimeUnit.MILLISECONDS); + + assertTrue(latch.await(10, TimeUnit.SECONDS)); + assertThat(latch.getCount(), is(0l)); Message reply = replyChannel.receive(2000); assertNotNull(reply); @@ -101,7 +104,7 @@ public class ConcurrentAggregatorTests { new AggregatorTestTask(this.aggregator, message1, latch).run(); new AggregatorTestTask(this.aggregator, message2, latch).run(); new AggregatorTestTask(this.aggregator, message3, latch).run(); - Message reply = replyChannel.receive(500); + Message reply = replyChannel.receive(1000); assertNotNull(reply); assertEquals("123456789", reply.getPayload()); } @@ -117,13 +120,15 @@ public class ConcurrentAggregatorTests { AggregatorTestTask task = new AggregatorTestTask(this.aggregator, message, latch); this.taskExecutor.execute(task); - latch.await(200, TimeUnit.MILLISECONDS); + + assertTrue(latch.await(10, TimeUnit.SECONDS)); + assertEquals("Task should have completed within timeout", 0, latch .getCount()); - Message reply = replyChannel.receive(100); + Message reply = replyChannel.receive(1000); assertNull("No message should have been sent normally", reply); this.store.expireMessageGroups(-10000); - Message discardedMessage = discardChannel.receive(100); + Message discardedMessage = discardChannel.receive(1000); assertNotNull("A message should have been discarded", discardedMessage); assertEquals(message, discardedMessage); } @@ -142,11 +147,13 @@ public class ConcurrentAggregatorTests { message2, latch); this.taskExecutor.execute(task1); this.taskExecutor.execute(task2); - latch.await(300, TimeUnit.MILLISECONDS); + + assertTrue(latch.await(10, TimeUnit.SECONDS)); + assertEquals("handlers should have been invoked within time limit", 0, latch.getCount()); this.store.expireMessageGroups(-10000); - Message reply = replyChannel.receive(100); + Message reply = replyChannel.receive(1000); assertNotNull("A reply message should have been received", reply); assertEquals(15, reply.getPayload()); assertNull(task1.getException()); @@ -179,13 +186,15 @@ public class ConcurrentAggregatorTests { message3, latch)); this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message4, latch)); - latch.await(1000, TimeUnit.MILLISECONDS); + + assertTrue(latch.await(10, TimeUnit.SECONDS)); + @SuppressWarnings("unchecked") - Message reply1 = (Message) replyChannel1.receive(500); + Message reply1 = (Message) replyChannel1.receive(1000); assertNotNull(reply1); assertThat(reply1.getPayload(), is(105)); @SuppressWarnings("unchecked") - Message reply2 = (Message) replyChannel2.receive(500); + Message reply2 = (Message) replyChannel2.receive(1000); assertNotNull(reply2); assertThat(reply2.getPayload(), is(2431)); } @@ -201,17 +210,17 @@ public class ConcurrentAggregatorTests { this.aggregator.setDiscardChannel(discardChannel); this.aggregator.handleMessage(createMessage(1, 1, 1, 1, replyChannel, null)); - assertEquals(1, replyChannel.receive(100).getPayload()); + assertEquals(1, replyChannel.receive(1000).getPayload()); this.aggregator.handleMessage(createMessage(3, 2, 1, 1, replyChannel, null)); - assertEquals(3, replyChannel.receive(100).getPayload()); + assertEquals(3, replyChannel.receive(1000).getPayload()); this.aggregator.handleMessage(createMessage(4, 3, 1, 1, replyChannel, null)); - assertEquals(4, replyChannel.receive(100).getPayload()); + assertEquals(4, replyChannel.receive(1000).getPayload()); // next message with same correlation ID is discarded this.aggregator.handleMessage(createMessage(2, 1, 1, 1, replyChannel, null)); - assertEquals(2, discardChannel.receive(100).getPayload()); + assertEquals(2, discardChannel.receive(1000).getPayload()); } @Test @@ -225,19 +234,19 @@ public class ConcurrentAggregatorTests { this.aggregator.setDiscardChannel(discardChannel); this.aggregator.handleMessage(createMessage(1, 1, 1, 1, replyChannel, null)); - assertEquals(1, replyChannel.receive(100).getPayload()); + assertEquals(1, replyChannel.receive(1000).getPayload()); this.aggregator.handleMessage(createMessage(2, 2, 1, 1, replyChannel, null)); - assertEquals(2, replyChannel.receive(100).getPayload()); + assertEquals(2, replyChannel.receive(1000).getPayload()); this.aggregator.handleMessage(createMessage(3, 3, 1, 1, replyChannel, null)); - assertEquals(3, replyChannel.receive(100).getPayload()); + assertEquals(3, replyChannel.receive(1000).getPayload()); this.aggregator.handleMessage(createMessage(4, 4, 1, 1, replyChannel, null)); - assertEquals(4, replyChannel.receive(100).getPayload()); + assertEquals(4, replyChannel.receive(1000).getPayload()); this.aggregator.handleMessage(createMessage(5, 1, 1, 1, replyChannel, null)); - assertEquals(5, replyChannel.receive(100).getPayload()); + assertEquals(5, replyChannel.receive(1000).getPayload()); assertNull(discardChannel.receive(0)); } @@ -266,8 +275,10 @@ public class ConcurrentAggregatorTests { message3, latch)); this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message4, latch)); - latch.await(1000, TimeUnit.MILLISECONDS); - Message reply = replyChannel.receive(100); + + assertTrue(latch.await(10, TimeUnit.SECONDS)); + + Message reply = replyChannel.receive(1000); assertNotNull("A message should be aggregated", reply); assertThat(((Integer) reply.getPayload()), is(105)); } @@ -290,11 +301,13 @@ public class ConcurrentAggregatorTests { AggregatorTestTask task3 = new AggregatorTestTask(aggregator, message3, latch); this.taskExecutor.execute(task3); - latch.await(1000, TimeUnit.MILLISECONDS); + + assertTrue(latch.await(10, TimeUnit.SECONDS)); + assertNull(task1.getException()); assertNull(task2.getException()); assertNull(task3.getException()); - Message reply = replyChannel.receive(500); + Message reply = replyChannel.receive(1000); assertNull(reply); } @@ -364,4 +377,4 @@ public class ConcurrentAggregatorTests { } } -} \ No newline at end of file +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java index 037c096ddf..8d20b2d71a 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java @@ -16,10 +16,19 @@ package org.springframework.integration.aggregator; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; +import static org.mockito.Matchers.isA; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + import java.util.ArrayList; import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import org.junit.Before; import org.junit.Test; @@ -36,14 +45,6 @@ import org.springframework.integration.store.SimpleMessageGroup; import org.springframework.integration.store.SimpleMessageStore; import org.springframework.integration.support.MessageBuilder; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.fail; - -import static org.mockito.Matchers.isA; -import static org.mockito.Mockito.doAnswer; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.when; - /** * @author Iwein Fuld * @author Dave Syer @@ -104,7 +105,7 @@ public class CorrelatingMessageHandlerTests { when(correlationStrategy.getCorrelationKey(isA(Message.class))).thenReturn(correlationKey); handler.setExpireGroupsUponCompletion(true); - + handler.handleMessage(message1); try { @@ -147,10 +148,9 @@ public class CorrelatingMessageHandlerTests { } }); - Thread.sleep(20); - assertEquals(0, store.expireMessageGroups(10000)); + assertTrue(bothMessagesHandled.await(10, TimeUnit.SECONDS)); - bothMessagesHandled.await(); + assertEquals(0, store.expireMessageGroups(10000)); } @Test diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/scenarios/AggregatorWithCustomReleaseStrategyTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/scenarios/AggregatorWithCustomReleaseStrategyTests.java index 0f459f5d26..cf2d455567 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/scenarios/AggregatorWithCustomReleaseStrategyTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/scenarios/AggregatorWithCustomReleaseStrategyTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 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 @@ -12,6 +12,9 @@ */ package org.springframework.integration.aggregator.scenarios; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -24,9 +27,6 @@ import org.springframework.integration.MessageChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.support.MessageBuilder; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertTrue; - /** * @author Oleg Zhurakousky * @@ -79,11 +79,11 @@ public class AggregatorWithCustomReleaseStrategyTests { assertTrue("Sends failed to complete", latch.await(10, TimeUnit.SECONDS)); - Message message = resultChannel.receive(10); + Message message = resultChannel.receive(1000); int counter = 0; while(message != null){ counter++; - message = resultChannel.receive(10); + message = resultChannel.receive(1000); } assertEquals(600, counter); } @@ -119,10 +119,10 @@ public class AggregatorWithCustomReleaseStrategyTests { assertTrue("Sends failed to complete", latch.await(10, TimeUnit.SECONDS)); - Message message = resultChannel.receive(10); + Message message = resultChannel.receive(1000); int counter = 0; while(message != null && ++counter < 7200){ - message = resultChannel.receive(10); + message = resultChannel.receive(1000); } assertEquals(7200, counter); }