From efcc01804cd3c0da1d853cfc6ae85728868e165e Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 25 Nov 2020 12:13:33 -0500 Subject: [PATCH] Some miscellaneous fixes Related to https://build.spring.io/browse/INT-MASTERSPRING40-JOB1-1271 * Fix race condition with subscription in the `FluxMessageChannel`: rely on the `doOnRequest()` to start producing from upstream publishers * Remove `SourcePollingChannelAdapterTests` since `Class.getDeclaredConstructor().newInstance()` doesn't rethrow an exception as is but wrap it into the `InvocationTargetException`. This is probably the main reason to deprecate a regular `Class.newInstance()` * Use `LettuceConnectionFactory.setEagerInitialization(true)` in the `RedisAvailableRule` to avoid possible dead lock with lazy init * Adjust Redis tests for new `RedisAvailableRule` behavior --- .../channel/FluxMessageChannel.java | 3 +- .../SourcePollingChannelAdapterTests.java | 116 ------------------ .../redis/config/RedisChannelParserTests.java | 6 +- ...boundChannelAdapterParserTests-context.xml | 3 +- ...RedisInboundChannelAdapterParserTests.java | 15 +-- ...boundChannelAdapterParserTests-context.xml | 7 +- .../redis/rules/RedisAvailableRule.java | 15 ++- 7 files changed, 30 insertions(+), 135 deletions(-) delete mode 100644 spring-integration-core/src/test/java/org/springframework/integration/endpoint/SourcePollingChannelAdapterTests.java 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 af2edff793..3f222e50ce 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 @@ -94,11 +94,10 @@ public class FluxMessageChannel extends AbstractMessageChannel @Override public void subscribe(Subscriber> subscriber) { this.sink.asFlux() + .doOnRequest((r) -> this.subscribedSignal.tryEmitNext(true)) .doFinally((s) -> this.subscribedSignal.tryEmitNext(this.sink.currentSubscriberCount() > 0)) .share() .subscribe(subscriber); - - this.subscribedSignal.tryEmitNext(this.sink.currentSubscriberCount() > 0); } @Override diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/SourcePollingChannelAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/SourcePollingChannelAdapterTests.java deleted file mode 100644 index 86bfa676eb..0000000000 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/SourcePollingChannelAdapterTests.java +++ /dev/null @@ -1,116 +0,0 @@ -/* - * Copyright 2020 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 - * - * https://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. - */ - -package org.springframework.integration.endpoint; - -import static org.assertj.core.api.Assertions.assertThat; - -import java.io.IOException; -import java.lang.reflect.UndeclaredThrowableException; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; - -import org.aopalliance.intercept.MethodInterceptor; -import org.junit.jupiter.api.Test; - -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; -import org.springframework.integration.config.EnableIntegration; -import org.springframework.integration.core.MessageSource; -import org.springframework.integration.dsl.IntegrationFlow; -import org.springframework.integration.dsl.IntegrationFlows; -import org.springframework.integration.dsl.Pollers; -import org.springframework.lang.Nullable; -import org.springframework.messaging.Message; -import org.springframework.test.annotation.DirtiesContext; -import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; - -/** - * @author Gary Russell - * @since 4.3.23 - * - */ -@SpringJUnitConfig -@DirtiesContext -public class SourcePollingChannelAdapterTests { - - @Autowired - Config config; - - @Test - void undeclaredCheckedException() throws InterruptedException { - assertThat(this.config.latchl.await(10, TimeUnit.SECONDS)).isTrue(); - assertThat(this.config.caught).isInstanceOf(UndeclaredThrowableException.class); - } - - @Configuration - @EnableIntegration - public static class Config { - - volatile Exception caught; - - volatile CountDownLatch latchl = new CountDownLatch(1); - - @Bean - public IntegrationFlow flow() { - return IntegrationFlows.from(new BadMessageSource(), e -> e.poller(Pollers.fixedDelay(500) - .advice(exceptionCaptor()))) - .nullChannel(); - } - - private MethodInterceptor exceptionCaptor() { - return invocation -> { - try { - return invocation.proceed(); - } - catch (Exception e) { - Config.this.caught = e; - throw e; - } - finally { - Config.this.latchl.countDown(); - } - }; - } - - } - - public static class BadMessageSource implements MessageSource { - - @Override - @Nullable - public Message receive() { - try { - Foo.class.newInstance(); - } - catch (InstantiationException | IllegalAccessException e) { - } - return null; - } - - } - - public static class Foo { - - public Foo() throws IOException { - throw new IOException(); - } - - } - -} - diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisChannelParserTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisChannelParserTests.java index a5cb0e9a96..fea945eba5 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisChannelParserTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisChannelParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2020 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. @@ -73,10 +73,12 @@ public class RedisChannelParserTests extends RedisAvailableTests { } @Test + @RedisAvailable public void testPubSubChannelConfig() { RedisConnectionFactory connectionFactory = TestUtils.getPropertyValue(this.redisChannel, "connectionFactory", RedisConnectionFactory.class); - RedisSerializer redisSerializer = TestUtils.getPropertyValue(redisChannel, "serializer", RedisSerializer.class); + RedisSerializer redisSerializer = TestUtils.getPropertyValue(redisChannel, "serializer", + RedisSerializer.class); assertThat(this.context.getBean("redisConnectionFactory")).isEqualTo(connectionFactory); assertThat(this.context.getBean("redisSerializer")).isEqualTo(redisSerializer); assertThat(TestUtils.getPropertyValue(redisChannel, "topicName")).isEqualTo("si.test.topic.parser"); diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests-context.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests-context.xml index fcda07f122..181d7757a9 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests-context.xml +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests-context.xml @@ -16,7 +16,8 @@ id="adapter" topics="foo" topic-patterns="f*, b*" channel="receiveChannel" error-channel="testErrorChannel" message-converter="testConverter" serializer="serializer" - task-executor="executor"/> + task-executor="executor" + auto-startup="false"/> diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests.java index 63efbbf62d..372e9bbc55 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2020 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,14 +32,14 @@ import org.springframework.data.redis.listener.RedisMessageListenerContainer; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.redis.inbound.RedisInboundChannelAdapter; import org.springframework.integration.redis.rules.RedisAvailable; +import org.springframework.integration.redis.rules.RedisAvailableRule; import org.springframework.integration.redis.rules.RedisAvailableTests; import org.springframework.integration.support.converter.SimpleMessageConverter; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.test.annotation.DirtiesContext; -import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.context.junit4.SpringRunner; /** * @author Oleg Zhurakousky @@ -47,9 +47,9 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; * @author Gary Russell * @author Gunnar Hillert * @author Venil Noronha + * @author Artem Bilan */ -@ContextConfiguration -@RunWith(SpringJUnit4ClassRunner.class) +@RunWith(SpringRunner.class) @DirtiesContext public class RedisInboundChannelAdapterParserTests extends RedisAvailableTests { @@ -91,10 +91,11 @@ public class RedisInboundChannelAdapterParserTests extends RedisAvailableTests { @RedisAvailable public void testInboundChannelAdapterMessaging() throws Exception { RedisInboundChannelAdapter adapter = context.getBean("adapter", RedisInboundChannelAdapter.class); - this.awaitContainerSubscribedWithPatterns(TestUtils.getPropertyValue(adapter, "container", + adapter.start(); + awaitContainerSubscribedWithPatterns(TestUtils.getPropertyValue(adapter, "container", RedisMessageListenerContainer.class)); - RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest(); + RedisConnectionFactory connectionFactory = RedisAvailableRule.connectionFactory; connectionFactory.getConnection().publish("foo".getBytes(), "Hello Redis from foo".getBytes()); connectionFactory.getConnection().publish("bar".getBytes(), "Hello Redis from bar".getBytes()); diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParserTests-context.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParserTests-context.xml index 660045f344..665dbb9d7a 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParserTests-context.xml +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParserTests-context.xml @@ -24,19 +24,20 @@ - + - + - + diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableRule.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableRule.java index 087332cd74..98af8cae92 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableRule.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableRule.java @@ -42,6 +42,8 @@ public final class RedisAvailableRule implements MethodRule { public static LettuceConnectionFactory connectionFactory; + private static volatile boolean initialized; + protected static void setupConnectionFactory() { RedisStandaloneConfiguration redisStandaloneConfiguration = new RedisStandaloneConfiguration(); redisStandaloneConfiguration.setPort(REDIS_PORT); @@ -59,29 +61,34 @@ public final class RedisAvailableRule implements MethodRule { .build(); connectionFactory = new LettuceConnectionFactory(redisStandaloneConfiguration, clientConfiguration); - connectionFactory.afterPropertiesSet(); + connectionFactory.setEagerInitialization(true); } public static void cleanUpConnectionFactoryIfAny() { - if (connectionFactory != null) { + if (initialized) { connectionFactory.destroy(); + initialized = false; } } public Statement apply(final Statement base, final FrameworkMethod method, Object target) { return new Statement() { + @Override public void evaluate() throws Throwable { RedisAvailable redisAvailable = method.getAnnotation(RedisAvailable.class); if (redisAvailable != null) { if (connectionFactory != null) { try { - connectionFactory.getConnection(); + connectionFactory.afterPropertiesSet(); + initialized = true; } catch (Exception e) { - Assume.assumeTrue("Skipping test due to Redis not being available on port: " + REDIS_PORT + ": " + e, false); + Assume.assumeTrue( + "Skipping test due to Redis not being available on port: " + REDIS_PORT + ": " + e, + false); } base.evaluate(); }