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

Shutdown test executors in -core

Browse Source 
- don't use `ExecutorService` as `@Bean` - spring can't stop them
- also add log adjuster for ftp test that sometimes fails
This commit is contained in:
Gary Russell
2018-01-18 17:37:17 -05:00
committed by Artem Bilan
parent d890f14c5d
commit 8aa91d1db0
20 changed files with 137 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java
View File

@@ -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.
@@ -77,7 +77,8 @@ public class AbstractCorrelatingMessageHandlerTests {
/*
* Runs "reap" when group 'bar' is in completion
*/
Executors.newSingleThreadExecutor().execute(() -> {
ExecutorService exec = Executors.newSingleThreadExecutor();
exec.execute(() -> {
try {
waitReapStartLatch.await(10, TimeUnit.SECONDS);
}
@@ -145,6 +146,7 @@ public class AbstractCorrelatingMessageHandlerTests {
assertEquals(1, ((MessageGroup) outputMessages.get(1).getPayload()).size()); // 'qux'
assertNull(discards.receive(0));
exec.shutdownNow();
}
@Test // INT-2833

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

@@ -272,7 +272,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(1000);
Message<?> reply = replyChannel.receive(0);
assertNull("No message should have been sent normally", reply);
Message<?> discardedMessage = discardChannel.receive(1000);
assertNotNull("A message should have been discarded", discardedMessage);

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

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2016 the original author or authors.
* Copyright 2015-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.
@@ -132,6 +132,7 @@ public class BarrierMessageHandlerTests {
assertEquals("bar", result.get(1));
assertEquals(0, suspensions.size());
assertEquals(0, inProcess.size());
exec.shutdownNow();
}
@Test
@@ -141,8 +142,8 @@ public class BarrierMessageHandlerTests {
handler.setOutputChannel(outputChannel);
handler.setBeanFactory(mock(BeanFactory.class));
handler.afterPropertiesSet();
Executors.newSingleThreadExecutor()
.execute(() -> handler.trigger(MessageBuilder.withPayload("bar").setCorrelationId("foo").build()));
ExecutorService exec = Executors.newSingleThreadExecutor();
exec.execute(() -> handler.trigger(MessageBuilder.withPayload("bar").setCorrelationId("foo").build()));
Map<?, ?> suspensions = TestUtils.getPropertyValue(handler, "suspensions", Map.class);
int n = 0;
while (n++ < 100 && suspensions.size() == 0) {
@@ -156,6 +157,7 @@ public class BarrierMessageHandlerTests {
assertEquals("foo", result.get(0));
assertEquals("bar", result.get(1));
assertEquals(0, suspensions.size());
exec.shutdownNow();
}
@Test
@@ -169,7 +171,8 @@ public class BarrierMessageHandlerTests {
handler.setBeanFactory(mock(BeanFactory.class));
handler.afterPropertiesSet();
final CountDownLatch latch = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(() -> {
ExecutorService exec = Executors.newSingleThreadExecutor();
exec.execute(() -> {
handler.handleMessage(MessageBuilder.withPayload("foo").setCorrelationId("foo").build());
latch.countDown();
});
@@ -190,6 +193,7 @@ public class BarrierMessageHandlerTests {
assertSame(discard, triggerMessage);
handler.handleMessage(MessageBuilder.withPayload("foo").setCorrelationId("foo").build());
assertEquals(0, suspensions.size());
exec.shutdownNow();
}
@Test
@@ -218,7 +222,8 @@ public class BarrierMessageHandlerTests {
handler.afterPropertiesSet();
final AtomicReference<Exception> exception = new AtomicReference<Exception>();
final CountDownLatch latch = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(() -> {
ExecutorService exec = Executors.newSingleThreadExecutor();
exec.execute(() -> {
try {
handler.handleMessage(MessageBuilder.withPayload("foo").setCorrelationId("foo").build());
}
@@ -238,6 +243,7 @@ public class BarrierMessageHandlerTests {
assertTrue(latch.await(10, TimeUnit.SECONDS));
assertSame(exc, exception.get().getCause());
assertEquals(0, suspensions.size());
exec.shutdownNow();
}
@Test

13
spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageBarrierTests.java
View File

@@ -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.
@@ -27,6 +27,7 @@ import static org.mockito.Mockito.when;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
@@ -89,8 +90,9 @@ public class CorrelatingMessageBarrierTests {
barrier.setReleaseStrategy(trackingReleaseStrategy);
final CountDownLatch start = new CountDownLatch(1);
final CountDownLatch sent = new CountDownLatch(200);
ExecutorService exec = Executors.newSingleThreadExecutor();
for (int i = 0; i < 200; i++) {
sendAsynchronously(barrier, testMessage(), start, sent);
sendAsynchronously(barrier, testMessage(), start, sent, exec);
}
start.countDown();
@@ -106,10 +108,12 @@ public class CorrelatingMessageBarrierTests {
trackingReleaseStrategy.release("foo");
assertThat((barrier.receive()), is(notNullValue()));
}
exec.shutdownNow();
}
private void sendAsynchronously(final MessageHandler handler, final Message<Object> message, final CountDownLatch start, final CountDownLatch sent) {
Executors.newSingleThreadExecutor().execute(() -> {
private void sendAsynchronously(final MessageHandler handler, final Message<Object> message,
final CountDownLatch start, final CountDownLatch sent, ExecutorService exec) {
exec.execute(() -> {
try {
start.await();
}
@@ -119,7 +123,6 @@ public class CorrelatingMessageBarrierTests {
handler.handleMessage(message);
sent.countDown();
});
}
private Message<Object> testMessage() {

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

@@ -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.
@@ -28,6 +28,7 @@ import static org.mockito.Mockito.when;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
@@ -146,7 +147,8 @@ public class CorrelatingMessageHandlerTests {
handler.handleMessage(message1);
bothMessagesHandled.countDown();
storedMessages.add(message1);
Executors.newSingleThreadExecutor().submit(() -> {
ExecutorService exec = Executors.newSingleThreadExecutor();
exec.submit(() -> {
handler.handleMessage(message2);
storedMessages.add(message2);
bothMessagesHandled.countDown();
@@ -155,6 +157,7 @@ public class CorrelatingMessageHandlerTests {
assertTrue(bothMessagesHandled.await(10, TimeUnit.SECONDS));
assertEquals(0, store.expireMessageGroups(10000));
exec.shutdownNow();
}
@Test

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

@@ -92,7 +92,7 @@ public class AggregatorWithCustomReleaseStrategyTests {
});
}
assertTrue("Sends failed to complete: " + latch.getCount() + " remain", latch.await(60, TimeUnit.SECONDS));
assertTrue("Sends failed to complete: " + latch.getCount() + " remain", latch.await(120, TimeUnit.SECONDS));
Message<?> message = resultChannel.receive(1000);
int counter = 0;

10
spring-integration-core/src/test/java/org/springframework/integration/channel/CGLibProxyChannelTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017 the original author or authors.
* Copyright 2017-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.
@@ -20,8 +20,7 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
@@ -36,6 +35,7 @@ import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit4.SpringRunner;
@@ -128,8 +128,8 @@ public class CGLibProxyChannelTests {
}
@Bean
public ExecutorService executor() {
return Executors.newCachedThreadPool();
public Executor executor() {
return new ThreadPoolTaskExecutor();
}
private ProxyFactoryBean createProxyFactory(MessageChannel channel) {

21
spring-integration-core/src/test/java/org/springframework/integration/channel/ExecutorChannelTests.java
View File

@@ -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.
@@ -32,6 +32,7 @@ import static org.mockito.Mockito.verify;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@@ -76,8 +77,9 @@ public class ExecutorChannelTests {
@Test
public void roundRobinLoadBalancing() throws Exception {
int numberOfMessages = 11;
ConcurrentTaskExecutor taskExecutor = new ConcurrentTaskExecutor(
Executors.newSingleThreadScheduledExecutor(new CustomizableThreadFactory("test-")));
ScheduledExecutorService exec = Executors
.newSingleThreadScheduledExecutor(new CustomizableThreadFactory("test-"));
ConcurrentTaskExecutor taskExecutor = new ConcurrentTaskExecutor(exec);
ExecutorChannel channel = new ExecutorChannel(
taskExecutor, new RoundRobinLoadBalancingStrategy());
CountDownLatch latch = new CountDownLatch(numberOfMessages);
@@ -104,13 +106,15 @@ public class ExecutorChannelTests {
assertEquals(4, handler1.count.get());
assertEquals(4, handler2.count.get());
assertEquals(3, handler3.count.get());
exec.shutdownNow();
}
@Test
public void verifyFailoverWithLoadBalancing() throws Exception {
int numberOfMessages = 11;
ConcurrentTaskExecutor taskExecutor = new ConcurrentTaskExecutor(
Executors.newSingleThreadScheduledExecutor(new CustomizableThreadFactory("test-")));
ScheduledExecutorService exec = Executors
.newSingleThreadScheduledExecutor(new CustomizableThreadFactory("test-"));
ConcurrentTaskExecutor taskExecutor = new ConcurrentTaskExecutor(exec);
ExecutorChannel channel = new ExecutorChannel(
taskExecutor, new RoundRobinLoadBalancingStrategy());
CountDownLatch latch = new CountDownLatch(numberOfMessages);
@@ -138,13 +142,15 @@ public class ExecutorChannelTests {
assertEquals(0, handler2.count.get());
assertEquals(4, handler1.count.get());
assertEquals(7, handler3.count.get());
exec.shutdownNow();
}
@Test
public void verifyFailoverWithoutLoadBalancing() throws Exception {
int numberOfMessages = 11;
ConcurrentTaskExecutor taskExecutor = new ConcurrentTaskExecutor(
Executors.newSingleThreadScheduledExecutor(new CustomizableThreadFactory("test-")));
ScheduledExecutorService exec = Executors
.newSingleThreadScheduledExecutor(new CustomizableThreadFactory("test-"));
ConcurrentTaskExecutor taskExecutor = new ConcurrentTaskExecutor(exec);
ExecutorChannel channel = new ExecutorChannel(taskExecutor, null);
CountDownLatch latch = new CountDownLatch(numberOfMessages);
TestHandler handler1 = new TestHandler(latch);
@@ -169,6 +175,7 @@ public class ExecutorChannelTests {
assertEquals(0, handler1.count.get());
assertEquals(0, handler3.count.get());
assertEquals(numberOfMessages, handler2.count.get());
exec.shutdownNow();
}
@Test

13
spring-integration-core/src/test/java/org/springframework/integration/channel/MixedDispatcherConfigurationScenarioTests.java
View File

@@ -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.
@@ -32,6 +32,7 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -40,7 +41,7 @@ import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.junit.MockitoJUnitRunner;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.MessageRejectedException;
import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy;
@@ -69,7 +70,7 @@ public class MixedDispatcherConfigurationScenarioTests {
@Mock
private List<Exception> exceptionRegistry;
private ApplicationContext ac;
private ConfigurableApplicationContext ac;
@Mock
private MessageHandler handlerA;
@@ -99,6 +100,12 @@ public class MixedDispatcherConfigurationScenarioTests {
failed = new AtomicBoolean(false);
}
@After
public void tearDown() {
this.executor.shutdownNow();
this.ac.close();
}
@Test
public void noFailoverNoLoadBalancing() {
DirectChannel channel = (DirectChannel) ac.getBean("noLoadBalancerNoFailover");

7
spring-integration-core/src/test/java/org/springframework/integration/channel/PriorityChannelTests.java
View File

@@ -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.
@@ -24,7 +24,6 @@ import static org.junit.Assert.assertTrue;
import java.util.Comparator;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
@@ -38,6 +37,7 @@ import org.springframework.messaging.support.GenericMessage;
/**
* @author Mark Fisher
* @author Gary Russell
*/
public class PriorityChannelTests {
@@ -246,7 +246,7 @@ public class PriorityChannelTests {
final PriorityChannel channel = new PriorityChannel(1);
final AtomicBoolean sentSecondMessage = new AtomicBoolean(false);
final CountDownLatch latch = new CountDownLatch(1);
Executor executor = Executors.newSingleThreadScheduledExecutor();
ExecutorService executor = Executors.newSingleThreadScheduledExecutor();
channel.send(new GenericMessage<String>("test-1"));
executor.execute(() -> {
sentSecondMessage.set(channel.send(new GenericMessage<String>("test-2"), 3000));
@@ -262,6 +262,7 @@ public class PriorityChannelTests {
Message<?> message2 = channel.receive();
assertNotNull(message2);
assertEquals("test-2", message2.getPayload());
executor.shutdownNow();
}
@Test

7
spring-integration-core/src/test/java/org/springframework/integration/channel/QueueChannelTests.java
View File

@@ -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.
@@ -23,7 +23,7 @@ import static org.junit.Assert.assertTrue;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -68,7 +68,7 @@ public class QueueChannelTests {
final QueueChannel channel = new QueueChannel();
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
Executor singleThreadExecutor = Executors.newSingleThreadExecutor();
ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
Runnable receiveTask1 = () -> {
Message<?> message = channel.receive(0);
if (message != null) {
@@ -91,6 +91,7 @@ public class QueueChannelTests {
singleThreadExecutor.execute(receiveTask2);
latch2.await();
assertTrue(messageReceived.get());
singleThreadExecutor.shutdownNow();
}
@Test

22
spring-integration-core/src/test/java/org/springframework/integration/dispatcher/RoundRobinDispatcherConcurrentTests.java
View File

@@ -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.
@@ -26,6 +26,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -49,7 +50,7 @@ public class RoundRobinDispatcherConcurrentTests {
private final UnicastingDispatcher dispatcher = new UnicastingDispatcher();
private final ThreadPoolTaskExecutor scheduler = new ThreadPoolTaskExecutor();
private final ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
@Mock
private MessageHandler handler1;
@@ -69,9 +70,14 @@ public class RoundRobinDispatcherConcurrentTests {
@Before
public void initialize() throws Exception {
dispatcher.setLoadBalancingStrategy(new RoundRobinLoadBalancingStrategy());
scheduler.setCorePoolSize(10);
scheduler.setMaxPoolSize(10);
scheduler.initialize();
executor.setCorePoolSize(10);
executor.setMaxPoolSize(10);
executor.initialize();
}
@After
public void tearDown() {
this.executor.shutdown();
}
@Test(timeout = 1000)
@@ -97,7 +103,7 @@ public class RoundRobinDispatcherConcurrentTests {
allDone.countDown();
};
for (int i = 0; i < TOTAL_EXECUTIONS; i++) {
scheduler.execute(messageSenderTask);
executor.execute(messageSenderTask);
}
start.countDown();
allDone.await();
@@ -131,7 +137,7 @@ public class RoundRobinDispatcherConcurrentTests {
allDone.countDown();
};
for (int i = 0; i < TOTAL_EXECUTIONS; i++) {
scheduler.execute(messageSenderTask);
executor.execute(messageSenderTask);
}
start.countDown();
allDone.await();
@@ -164,7 +170,7 @@ public class RoundRobinDispatcherConcurrentTests {
}
};
for (int i = 0; i < TOTAL_EXECUTIONS; i++) {
scheduler.execute(messageSenderTask);
executor.execute(messageSenderTask);
}
start.countDown();
allDone.await(5000, TimeUnit.MILLISECONDS);

8
spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2017 the original author or authors.
* Copyright 2016-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.
@@ -27,7 +27,6 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;
import org.junit.Test;
@@ -51,6 +50,7 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit4.SpringRunner;
@@ -157,7 +157,9 @@ public class CorrelationHandlerTests {
@Bean
public Executor taskExecutor() {
return Executors.newCachedThreadPool();
ThreadPoolTaskExecutor tpte = new ThreadPoolTaskExecutor();
tpte.setCorePoolSize(50);
return tpte;
}
@Bean

14
spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2017 the original author or authors.
* Copyright 2016-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,7 +28,7 @@ import static org.junit.Assert.fail;
import java.io.Serializable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
@@ -86,6 +86,7 @@ import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;
@@ -614,7 +615,7 @@ public class IntegrationFlowTests {
@Bean
public IntegrationFlow subscribersFlow() {
return flow -> flow
.publishSubscribeChannel(Executors.newCachedThreadPool(), s -> s
.publishSubscribeChannel(executor(), s -> s
.subscribe(f -> f
.<Integer>handle((p, h) -> p / 2)
.channel(MessageChannels.queue("subscriber1Results")))
@@ -625,6 +626,13 @@ public class IntegrationFlowTests {
.channel(MessageChannels.queue("subscriber3Results"));
}
@Bean
public Executor executor() {
ThreadPoolTaskExecutor tpte = new ThreadPoolTaskExecutor();
tpte.setCorePoolSize(50);
return tpte;
}
@Bean
public IntegrationFlow wireTapFlow1() {
return IntegrationFlows.from("tappedChannel1")

8
spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2017 the original author or authors.
* Copyright 2016-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.
@@ -27,6 +27,7 @@ import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
@@ -60,6 +61,7 @@ import reactor.core.publisher.Flux;
/**
* @author Artem Bilan
* @author Gary Russell
*
* @since 5.0
*/
@@ -116,8 +118,9 @@ public class ReactiveStreamsTests {
.doOnNext(p -> latch.countDown())
.subscribe();
ExecutorService exec = Executors.newSingleThreadExecutor();
Future<List<Integer>> future =
Executors.newSingleThreadExecutor().submit(() ->
exec.submit(() ->
Flux.just("11,12,13")
.map(v -> v.split(","))
.flatMapIterable(Arrays::asList)
@@ -138,6 +141,7 @@ public class ReactiveStreamsTests {
assertNotNull(integers);
assertEquals(7, integers.size());
exec.shutdownNow();
}
@Test

7
spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayProxyFactoryBeanTests.java
View File

@@ -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.
@@ -31,7 +31,7 @@ import java.util.Collections;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
@@ -214,7 +214,7 @@ public class GatewayProxyFactoryBeanTests {
final TestService service = (TestService) context.getBean("proxy");
final String[] results = new String[numRequests];
final CountDownLatch latch = new CountDownLatch(numRequests);
Executor executor = Executors.newFixedThreadPool(numRequests);
ExecutorService executor = Executors.newFixedThreadPool(numRequests);
for (int i = 0; i < numRequests; i++) {
final int count = i;
executor.execute(() -> {
@@ -237,6 +237,7 @@ public class GatewayProxyFactoryBeanTests {
assertEquals(numRequests, interceptor.getSentCount());
assertEquals(numRequests, interceptor.getReceivedCount());
context.close();
executor.shutdownNow();
}
@Test

12
spring-integration-core/src/test/java/org/springframework/integration/handler/AsyncHandlerTests.java
View File

@@ -32,10 +32,12 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.apache.commons.logging.Log;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@@ -75,14 +77,17 @@ public class AsyncHandlerTests {
private volatile CountDownLatch exceptionLatch = new CountDownLatch(1);
private ExecutorService executor;
@Before
public void setup() {
this.executor = Executors.newSingleThreadExecutor();
this.handler = new AbstractReplyProducingMessageHandler() {
@Override
protected Object handleRequestMessage(Message<?> requestMessage) {
final SettableListenableFuture<String> future = new SettableListenableFuture<String>();
Executors.newSingleThreadExecutor().execute(() -> {
AsyncHandlerTests.this.executor.execute(() -> {
try {
latch.await(10, TimeUnit.SECONDS);
switch (whichTest) {
@@ -118,6 +123,11 @@ public class AsyncHandlerTests {
}).when(logger).error(anyString(), any(Throwable.class));
}
@After
public void tearDown() {
this.executor.shutdownNow();
}
@Test
public void testGoodResult() {
this.whichTest = 0;

7
spring-integration-core/src/test/java/org/springframework/integration/handler/advice/AdvisedMessageHandlerTests.java
View File

@@ -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.
@@ -39,6 +39,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
@@ -890,7 +891,8 @@ public class AdvisedMessageHandlerTests {
PollableChannel inputChannel = new QueueChannel();
PollingConsumer consumer = new PollingConsumer(inputChannel, message -> { });
consumer.setAdviceChain(Collections.singletonList(advice));
consumer.setTaskExecutor(new ErrorHandlingTaskExecutor(Executors.newSingleThreadExecutor(), t -> { }));
ExecutorService exec = Executors.newSingleThreadExecutor();
consumer.setTaskExecutor(new ErrorHandlingTaskExecutor(exec, t -> { }));
consumer.setBeanFactory(mock(BeanFactory.class));
consumer.afterPropertiesSet();
consumer.setTaskScheduler(mock(TaskScheduler.class));
@@ -916,6 +918,7 @@ public class AdvisedMessageHandlerTests {
"an attempt to advise method 'call' in " +
"'org.springframework.integration.endpoint.AbstractPollingEndpoint"));
consumer.stop();
exec.shutdownNow();
}
public void filterDiscardNoAdvice() {

11
spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java
View File

@@ -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.
@@ -30,6 +30,7 @@ import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
@@ -107,7 +108,8 @@ public class SimpleMessageStoreTests {
final CountDownLatch message2Latch = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(() -> {
ExecutorService exec = Executors.newSingleThreadExecutor();
exec.execute(() -> {
store2.addMessage(testMessage2);
message2Latch.countDown();
});
@@ -119,6 +121,7 @@ public class SimpleMessageStoreTests {
assertTrue(message2Latch.await(10, TimeUnit.SECONDS));
Message<?> t2 = store2.getMessage(testMessage2.getHeaders().getId());
assertEquals(testMessage2, t2);
exec.shutdownNow();
}
@Test(expected = MessagingException.class)
@@ -150,7 +153,8 @@ public class SimpleMessageStoreTests {
final CountDownLatch message2Latch = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(() -> {
ExecutorService exec = Executors.newSingleThreadExecutor();
exec.execute(() -> {
store2.addMessageToGroup("foo", testMessage2);
message2Latch.countDown();
});
@@ -161,6 +165,7 @@ public class SimpleMessageStoreTests {
assertTrue(message2Latch.await(10, TimeUnit.SECONDS));
MessageGroup messageGroup = store2.getMessageGroup("foo");
messageGroup.getMessages().contains(testMessage2);
exec.shutdownNow();
}
@Test(expected = MessagingException.class)

6
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java
View File

@@ -26,6 +26,7 @@ import java.io.InputStream;
import java.util.Comparator;
import org.apache.commons.net.ftp.FTPFile;
import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -46,6 +47,7 @@ import org.springframework.integration.ftp.FtpTestSupport;
import org.springframework.integration.ftp.session.FtpFileInfo;
import org.springframework.integration.ftp.session.FtpRemoteFileTemplate;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.integration.test.rule.Log4j2LevelAdjuster;
import org.springframework.integration.transformer.StreamTransformer;
import org.springframework.messaging.Message;
import org.springframework.scheduling.support.PeriodicTrigger;
@@ -71,6 +73,10 @@ public class FtpStreamingMessageSourceTests extends FtpTestSupport {
@Autowired
private SourcePollingChannelAdapter adapter;
@Rule
public Log4j2LevelAdjuster adjuster = Log4j2LevelAdjuster.debug()
.categories("org.springframework.integration", "org.apache.commons");
@SuppressWarnings("unchecked")
@Test
public void testAllContents() {