Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-3171: Polishing Aggregator tests

Browse Source 
* Increase waiting timeouts
* Get rid of `Thread.sleep` when it's dangerous

JIRA: https://jira.springsource.org/browse/INT-3171
This commit is contained in:
Artem Bilan
2013-10-14 17:56:25 +03:00
committed by Gary Russell
parent 473a2282e9
commit 06979d7678
4 changed files with 92 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java
View File

@@ -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<Integer> reply1 = (Message<Integer>) replyChannel1.receive(500);
Message<Integer> reply1 = (Message<Integer>) replyChannel1.receive(1000);
assertNotNull(reply1);
assertThat(reply1.getPayload(), is(105));
@SuppressWarnings("unchecked")
Message<Integer> reply2 = (Message<Integer>) replyChannel2.receive(500);
Message<Integer> reply2 = (Message<Integer>) 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);
}

77
spring-integration-core/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java
View File

@@ -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<Integer> reply1 = (Message<Integer>) replyChannel1.receive(500);
Message<Integer> reply1 = (Message<Integer>) replyChannel1.receive(1000);
assertNotNull(reply1);
assertThat(reply1.getPayload(), is(105));
@SuppressWarnings("unchecked")
Message<Integer> reply2 = (Message<Integer>) replyChannel2.receive(500);
Message<Integer> reply2 = (Message<Integer>) 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 {
}
}
}
}

24
spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java
View File

@@ -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

16
spring-integration-core/src/test/java/org/springframework/integration/aggregator/scenarios/AggregatorWithCustomReleaseStrategyTests.java
View File

@@ -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);
}