diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java index 90f41694d8..2c53d14371 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2020 the original author or authors. + * Copyright 2015-2021 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 reactor.core.Disposables; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.core.publisher.Sinks; +import reactor.core.scheduler.Scheduler; import reactor.core.scheduler.Schedulers; /** @@ -47,6 +48,8 @@ import reactor.core.scheduler.Schedulers; public class FluxMessageChannel extends AbstractMessageChannel implements Publisher>, ReactiveStreamsSubscribableChannel { + private final Scheduler scheduler = Schedulers.boundedElastic(); + private final Sinks.Many> sink = Sinks.many().multicast().onBackpressureBuffer(1, false); private final Sinks.Many subscribedSignal = Sinks.many().replay().limit(1); @@ -114,7 +117,7 @@ public class FluxMessageChannel extends AbstractMessageChannel this.upstreamSubscriptions.add( Flux.from(publisher) .delaySubscription(this.subscribedSignal.asFlux().filter(Boolean::booleanValue).next()) - .publishOn(Schedulers.boundedElastic()) + .publishOn(this.scheduler) .doOnNext((message) -> { try { if (!send(message)) { @@ -135,6 +138,7 @@ public class FluxMessageChannel extends AbstractMessageChannel this.upstreamSubscriptions.dispose(); this.subscribedSignal.emitComplete(Sinks.EmitFailureHandler.FAIL_FAST); this.sink.emitComplete(Sinks.EmitFailureHandler.FAIL_FAST); + this.scheduler.dispose(); super.destroy(); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aop/PublisherAnnotationAdvisorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aop/PublisherAnnotationAdvisorTests.java index 195d270ba0..07f62a0295 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aop/PublisherAnnotationAdvisorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aop/PublisherAnnotationAdvisorTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2021 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,8 +23,9 @@ import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; -import org.junit.Before; -import org.junit.Test; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; import org.springframework.aop.framework.ProxyFactory; import org.springframework.context.support.StaticApplicationContext; @@ -36,6 +37,7 @@ import org.springframework.messaging.handler.annotation.Payload; /** * @author Mark Fisher * @author Jeff Maxwell + * @author Artem Bilan * * @since 2.0 */ @@ -43,12 +45,17 @@ public class PublisherAnnotationAdvisorTests { private final StaticApplicationContext context = new StaticApplicationContext(); - @Before + @BeforeEach public void setup() { context.registerSingleton("testChannel", QueueChannel.class); context.registerSingleton("testMetaChannel", QueueChannel.class); } + @AfterEach + void tearDown() { + this.context.close(); + } + @Test public void annotationAtMethodLevelOnVoidReturnWithParamAnnotation() { PublisherAnnotationAdvisor advisor = new PublisherAnnotationAdvisor(); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/DispatchingChannelErrorHandlingTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/DispatchingChannelErrorHandlingTests.java index 52d6a8c0ea..b5884cf1e4 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/DispatchingChannelErrorHandlingTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/DispatchingChannelErrorHandlingTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2021 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,11 +17,12 @@ package org.springframework.integration.channel; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; -import org.junit.Test; +import org.junit.jupiter.api.Test; import org.springframework.context.support.StaticApplicationContext; import org.springframework.core.task.SimpleAsyncTaskExecutor; @@ -35,21 +36,23 @@ import org.springframework.messaging.MessagingException; /** * @author Mark Fisher + * @author Artem Bilan + * * @since 1.0.3 */ public class DispatchingChannelErrorHandlingTests { private final CountDownLatch latch = new CountDownLatch(1); - - @Test(expected = MessageDeliveryException.class) + @Test public void handlerThrowsExceptionPublishSubscribeWithoutExecutor() { PublishSubscribeChannel channel = new PublishSubscribeChannel(); channel.subscribe(message -> { throw new UnsupportedOperationException("intentional test failure"); }); Message message = MessageBuilder.withPayload("test").build(); - channel.send(message); + assertThatExceptionOfType(MessageDeliveryException.class) + .isThrownBy(() -> channel.send(message)); } @Test @@ -72,13 +75,14 @@ public class DispatchingChannelErrorHandlingTests { }); Message message = MessageBuilder.withPayload("test").build(); channel.send(message); - this.waitForLatch(10000); + waitForLatch(10000); Message errorMessage = resultHandler.lastMessage; assertThat(errorMessage.getPayload().getClass()).isEqualTo(MessagingException.class); MessagingException exceptionPayload = (MessagingException) errorMessage.getPayload(); assertThat(exceptionPayload.getCause().getClass()).isEqualTo(UnsupportedOperationException.class); assertThat(exceptionPayload.getFailedMessage()).isSameAs(message); assertThat(resultHandler.lastThread).isNotSameAs(Thread.currentThread()); + context.close(); } @Test @@ -108,6 +112,7 @@ public class DispatchingChannelErrorHandlingTests { assertThat(exceptionPayload.getCause().getClass()).isEqualTo(UnsupportedOperationException.class); assertThat(exceptionPayload.getFailedMessage()).isSameAs(message); assertThat(resultHandler.lastThread).isNotSameAs(Thread.currentThread()); + context.close(); } @@ -136,8 +141,8 @@ public class DispatchingChannelErrorHandlingTests { this.lastThread = Thread.currentThread(); latch.countDown(); } - } + } @SuppressWarnings("serial") private static class TestTimedOutException extends RuntimeException { @@ -145,6 +150,7 @@ public class DispatchingChannelErrorHandlingTests { TestTimedOutException() { super("timed out while waiting for latch"); } + } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/MessageChannelReactiveUtilsTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/MessageChannelReactiveUtilsTests.java index 45436f3a33..c695d6ffc3 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/MessageChannelReactiveUtilsTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/MessageChannelReactiveUtilsTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2019-2020 the original author or authors. + * Copyright 2019-2021 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,6 +23,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.Test; import org.springframework.integration.util.IntegrationReactiveUtils; @@ -31,6 +32,7 @@ import org.springframework.messaging.support.GenericMessage; import reactor.core.Disposable; import reactor.core.Disposables; import reactor.core.publisher.Mono; +import reactor.core.scheduler.Scheduler; import reactor.core.scheduler.Schedulers; import reactor.test.StepVerifier; import reactor.util.concurrent.Queues; @@ -43,6 +45,13 @@ import reactor.util.concurrent.Queues; */ class MessageChannelReactiveUtilsTests { + private static final Scheduler SCHEDULER = Schedulers.boundedElastic(); + + @AfterAll + static void tearDown() { + SCHEDULER.dispose(); + } + @Test void testBackpressureWithSubscribableChannel() { Disposable.Composite compositeDisposable = Disposables.composite(); @@ -53,7 +62,7 @@ class MessageChannelReactiveUtilsTests { .expectSubscription() .then(() -> { compositeDisposable.add( - Schedulers.boundedElastic().schedule(() -> { + SCHEDULER.schedule(() -> { while (true) { if (channel.getSubscriberCount() > 0) { channel.send(new GenericMessage<>("foo")); @@ -84,7 +93,7 @@ class MessageChannelReactiveUtilsTests { .expectSubscription() .then(() -> compositeDisposable.add( - Schedulers.boundedElastic().schedule(() -> { + SCHEDULER.schedule(() -> { while (true) { if (channel.getSubscriberCount() > 0) { channel.send(new GenericMessage<>("foo")); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/MixedDispatcherConfigurationScenarioTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/channel/MixedDispatcherConfigurationScenarioTests-context.xml index 4c1c07a8f4..4bd953eadf 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/MixedDispatcherConfigurationScenarioTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/MixedDispatcherConfigurationScenarioTests-context.xml @@ -18,6 +18,7 @@ + @@ -25,6 +26,7 @@ + @@ -32,11 +34,13 @@ + - + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/MixedDispatcherConfigurationScenarioTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/MixedDispatcherConfigurationScenarioTests.java index 80f1d84994..edeb064a17 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/MixedDispatcherConfigurationScenarioTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/MixedDispatcherConfigurationScenarioTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2021 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. @@ -21,88 +21,72 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import java.util.List; import java.util.concurrent.CountDownLatch; -import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executor; 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; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; import org.mockito.InOrder; -import org.mockito.Mock; import org.mockito.Mockito; -import org.mockito.junit.MockitoJUnitRunner; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.ConfigurableApplicationContext; -import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.MessageRejectedException; import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy; import org.springframework.integration.dispatcher.UnicastingDispatcher; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.support.GenericMessage; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; /** * @author Oleg Zhurakousky * @author Gary Russell + * @author Artem Bilan */ -@RunWith(MockitoJUnitRunner.class) +@SpringJUnitConfig +@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD) public class MixedDispatcherConfigurationScenarioTests { private static final int TOTAL_EXECUTIONS = 40; - private ExecutorService executor; + private final CountDownLatch allDone = new CountDownLatch(TOTAL_EXECUTIONS); - private CountDownLatch allDone; + private final CountDownLatch start = new CountDownLatch(1); - private CountDownLatch start; - - private AtomicBoolean failed; - - @Mock - private List exceptionRegistry; - - private ConfigurableApplicationContext ac; - - @Mock - private MessageHandler handlerA; - - @Mock - private MessageHandler handlerB; - - @Mock - private MessageHandler handlerC; - - private final Message message = new GenericMessage("test"); + private final AtomicBoolean failed = new AtomicBoolean(false); + private final Message message = new GenericMessage<>("test"); @SuppressWarnings("unchecked") - @Before - public void initialize() throws Exception { - Mockito.reset(exceptionRegistry); - Mockito.reset(handlerA); - Mockito.reset(handlerB); - Mockito.reset(handlerC); + private final List exceptionRegistry = mock(List.class); - ac = new ClassPathXmlApplicationContext("MixedDispatcherConfigurationScenarioTests-context.xml", - MixedDispatcherConfigurationScenarioTests.class); - executor = ac.getBean("taskExecutor", ExecutorService.class); - allDone = new CountDownLatch(TOTAL_EXECUTIONS); - start = new CountDownLatch(1); - failed = new AtomicBoolean(false); - } + private final MessageHandler handlerA = mock(MessageHandler.class); - @After - public void tearDown() { - this.executor.shutdownNow(); - this.ac.close(); + private final MessageHandler handlerB = mock(MessageHandler.class); + + private final MessageHandler handlerC = mock(MessageHandler.class); + + @Autowired + @Qualifier("taskExecutor") + private Executor executor; + + @Autowired + private ConfigurableApplicationContext ac; + + @BeforeEach + public void initialize() { + Mockito.reset(this.exceptionRegistry, this.handlerA, this.handlerB, this.handlerC); } @Test @@ -159,9 +143,6 @@ public class MixedDispatcherConfigurationScenarioTests { start.countDown(); assertThat(allDone.await(10, TimeUnit.SECONDS)).isTrue(); - executor.shutdown(); - executor.awaitTermination(10, TimeUnit.SECONDS); - assertThat(failed.get()).as("not all messages were accepted").isTrue(); verify(handlerA, times(TOTAL_EXECUTIONS)).handleMessage(message); verify(handlerB, times(0)).handleMessage(message); @@ -169,8 +150,7 @@ public class MixedDispatcherConfigurationScenarioTests { } @Test - public void noFailoverNoLoadBalancingWithExecutorConcurrent() - throws Exception { + public void noFailoverNoLoadBalancingWithExecutorConcurrent() throws Exception { final ExecutorChannel channel = (ExecutorChannel) ac.getBean("noLoadBalancerNoFailoverExecutor"); UnicastingDispatcher dispatcher = channel.getDispatcher(); dispatcher.addHandler(handlerA); @@ -200,9 +180,6 @@ public class MixedDispatcherConfigurationScenarioTests { start.countDown(); assertThat(allDone.await(10, TimeUnit.SECONDS)).isTrue(); - executor.shutdown(); - executor.awaitTermination(10, TimeUnit.SECONDS); - assertThat(failed.get()).as("not all messages were accepted").isTrue(); verify(handlerA, times(TOTAL_EXECUTIONS)).handleMessage(message); verify(handlerB, times(0)).handleMessage(message); @@ -281,9 +258,6 @@ public class MixedDispatcherConfigurationScenarioTests { start.countDown(); assertThat(allDone.await(10, TimeUnit.SECONDS)).isTrue(); - executor.shutdown(); - executor.awaitTermination(10, TimeUnit.SECONDS); - assertThat(failed.get()).as("not all messages were accepted").isTrue(); verify(handlerA, times(14)).handleMessage(message); verify(handlerB, times(13)).handleMessage(message); @@ -334,9 +308,6 @@ public class MixedDispatcherConfigurationScenarioTests { start.countDown(); assertThat(allDone.await(10, TimeUnit.SECONDS)).isTrue(); - executor.shutdown(); - executor.awaitTermination(10, TimeUnit.SECONDS); - assertThat(failed.get()).as("not all messages were accepted").isTrue(); verify(handlerA, times(14)).handleMessage(message); verify(handlerB, times(13)).handleMessage(message); @@ -346,8 +317,7 @@ public class MixedDispatcherConfigurationScenarioTests { @Test public void failoverNoLoadBalancing() { - DirectChannel channel = (DirectChannel) ac - .getBean("noLoadBalancerFailover"); + DirectChannel channel = ac.getBean("noLoadBalancerFailover", DirectChannel.class); doThrow(new MessageRejectedException(message, null)).when(handlerA) .handleMessage(message); UnicastingDispatcher dispatcher = channel.getDispatcher(); @@ -376,10 +346,8 @@ public class MixedDispatcherConfigurationScenarioTests { } @Test - public void failoverNoLoadBalancingConcurrent() - throws Exception { - final DirectChannel channel = (DirectChannel) ac - .getBean("noLoadBalancerFailover"); + public void failoverNoLoadBalancingConcurrent() throws Exception { + final DirectChannel channel = ac.getBean("noLoadBalancerFailover", DirectChannel.class); doThrow(new MessageRejectedException(message, null)).when(handlerA).handleMessage(message); UnicastingDispatcher dispatcher = channel.getDispatcher(); dispatcher.addHandler(handlerA); @@ -415,9 +383,6 @@ public class MixedDispatcherConfigurationScenarioTests { start.countDown(); assertThat(allDone.await(10, TimeUnit.SECONDS)).isTrue(); - executor.shutdown(); - executor.awaitTermination(10, TimeUnit.SECONDS); - assertThat(failed.get()).as("not all messages were accepted").isFalse(); verify(handlerA, times(TOTAL_EXECUTIONS)).handleMessage(message); verify(handlerB, times(TOTAL_EXECUTIONS)).handleMessage(message); @@ -459,9 +424,6 @@ public class MixedDispatcherConfigurationScenarioTests { start.countDown(); assertThat(allDone.await(10, TimeUnit.SECONDS)).isTrue(); - executor.shutdown(); - executor.awaitTermination(10, TimeUnit.SECONDS); - verify(handlerA, times(TOTAL_EXECUTIONS)).handleMessage(message); verify(handlerB, times(TOTAL_EXECUTIONS)).handleMessage(message); verify(handlerC, never()).handleMessage(message); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/QueueChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/QueueChannelTests.java index bda1976cc5..c990f38ec9 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/QueueChannelTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/QueueChannelTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2021 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. @@ -29,9 +29,7 @@ import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; -import org.junit.Rule; -import org.junit.Test; -import org.junit.rules.TemporaryFolder; +import org.junit.jupiter.api.Test; import org.springframework.integration.selector.UnexpiredMessageSelector; import org.springframework.integration.support.MessageBuilder; @@ -286,121 +284,4 @@ public class QueueChannelTests { assertThat(channel.send(new GenericMessage<>("roomAvailable"), 0)).isTrue(); } - @Rule - public final TemporaryFolder tempFolder = new TemporaryFolder(); - - /*TODO: No Reactor Chronicle artifact - @Test - public void testReactorPersistentQueue() throws InterruptedException, IOException { - final AtomicBoolean messageReceived = new AtomicBoolean(false); - final CountDownLatch latch = new CountDownLatch(1); - PersistentQueue> queue = new PersistentQueueSpec>() - .codec(new JavaSerializationCodec>()) - .basePath(this.tempFolder.getRoot().getAbsolutePath()) - .get(); - - final QueueChannel channel = new QueueChannel(queue); - new Thread(new Runnable() { - @Override - public void run() { - Message message = channel.receive(); - if (message != null) { - messageReceived.set(true); - latch.countDown(); - } - } - }).start(); - assertFalse(messageReceived.get()); - channel.send(new GenericMessage("testing")); - latch.await(1000, TimeUnit.MILLISECONDS); - assertTrue(messageReceived.get()); - - final CountDownLatch latch1 = new CountDownLatch(2); - - Thread thread = new Thread(new Runnable() { - @Override - public void run() { - while (true) { - Message message = channel.receive(100); - if (message != null) { - latch1.countDown(); - if (latch1.getCount() == 0) { - break; - } - } - } - } - }); - thread.start(); - - Thread.sleep(200); - channel.send(new GenericMessage("testing")); - channel.send(new GenericMessage("testing")); - assertTrue(latch1.await(1000, TimeUnit.MILLISECONDS)); - - final AtomicBoolean receiveInterrupted = new AtomicBoolean(false); - final CountDownLatch latch2 = new CountDownLatch(1); - Thread t = new Thread(new Runnable() { - @Override - public void run() { - Message message = channel.receive(10000); - receiveInterrupted.set(true); - assertTrue(message == null); - latch2.countDown(); - } - }); - t.start(); - assertFalse(receiveInterrupted.get()); - t.interrupt(); - latch2.await(); - assertTrue(receiveInterrupted.get()); - - receiveInterrupted.set(false); - final CountDownLatch latch3 = new CountDownLatch(1); - t = new Thread(new Runnable() { - @Override - public void run() { - Message message = channel.receive(); - receiveInterrupted.set(true); - assertTrue(message == null); - latch3.countDown(); - } - }); - t.start(); - assertFalse(receiveInterrupted.get()); - t.interrupt(); - latch3.await(); - assertTrue(receiveInterrupted.get()); - - GenericMessage message1 = new GenericMessage("test1"); - GenericMessage message2 = new GenericMessage("test2"); - assertTrue(channel.send(message1)); - assertTrue(channel.send(message2)); - List> clearedMessages = channel.clear(); - assertNotNull(clearedMessages); - assertEquals(2, clearedMessages.size()); - - clearedMessages = channel.clear(); - assertNotNull(clearedMessages); - assertEquals(0, clearedMessages.size()); - - // Test on artificial infinite wait - // channel.receive(); - - // Distributed scenario - final CountDownLatch latch4 = new CountDownLatch(1); - new Thread(new Runnable() { - @Override - public void run() { - Message message = channel.receive(); - if (message != null) { - latch4.countDown(); - } - } - }).start(); - queue.add(new GenericMessage("foo")); - assertTrue(latch4.await(1000, TimeUnit.MILLISECONDS)); - } -*/ - } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java index 109bddd8a4..4327662736 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2020 the original author or authors. + * Copyright 2016-2021 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. @@ -110,6 +110,7 @@ public class ReactiveStreamsConsumerTests { assertThat(result).containsExactly(testMessage, testMessage2); reactiveConsumer.stop(); + testChannel.destroy(); } @@ -291,6 +292,7 @@ public class ReactiveStreamsConsumerTests { assertThat(result).containsExactly(testMessage, testMessage2, testMessage2); endpointFactoryBean.stop(); + testChannel.destroy(); } @Test @@ -331,6 +333,7 @@ public class ReactiveStreamsConsumerTests { .verify(Duration.ofSeconds(10)); reactiveConsumer.stop(); + testChannel.destroy(); } }