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

Some miscellaneous fixes

Browse Source 
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
This commit is contained in:
Artem Bilan
2020-11-25 12:13:33 -05:00
parent 63bd3e4938
commit efcc01804c
7 changed files with 30 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java
View File

@@ -94,11 +94,10 @@ public class FluxMessageChannel extends AbstractMessageChannel
@Override
public void subscribe(Subscriber<? super Message<?>> 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

116
spring-integration-core/src/test/java/org/springframework/integration/endpoint/SourcePollingChannelAdapterTests.java
View File

@@ -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<String> {
@Override
@Nullable
public Message<String> receive() {
try {
Foo.class.newInstance();
}
catch (InstantiationException | IllegalAccessException e) {
}
return null;
}
}
public static class Foo {
public Foo() throws IOException {
throw new IOException();
}
}
}

6
spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisChannelParserTests.java
View File

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

3
spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests-context.xml
View File

@@ -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"/>
<bean id="executor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
<property name="corePoolSize" value="5"/>

15
spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests.java
View File

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

7
spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParserTests-context.xml
View File

@@ -24,19 +24,20 @@
</int-redis:request-handler-advice-chain>
</int-redis:outbound-channel-adapter>
<int-redis:inbound-channel-adapter id="fooInbound" channel="receiveChannel" topics="foo"/>
<int-redis:inbound-channel-adapter id="fooInbound" channel="receiveChannel" topics="foo" auto-startup="false"/>
<int:channel id="receiveChannel">
<int:queue/>
</int:channel>
<int-redis:inbound-channel-adapter id="barInbound" channel="barChannel" topics="bar"/>
<int-redis:inbound-channel-adapter id="barInbound" channel="barChannel" topics="bar" auto-startup="false"/>
<int:channel id="barChannel">
<int:queue/>
</int:channel>
<bean id="testConverter" class="org.springframework.integration.redis.config.RedisOutboundChannelAdapterParserTests$TestMessageConverter"/>
<bean id="testConverter"
class="org.springframework.integration.redis.config.RedisOutboundChannelAdapterParserTests$TestMessageConverter"/>
<int:chain input-channel="redisOutboundChain">
<int-redis:outbound-channel-adapter topic="foo"/>

15
spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableRule.java
View File

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