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

Fix ReactiveStreamsConsumerTests and Checkstyle

Browse Source 
Also increase timeouts in the `RedisAvailableRule`
This commit is contained in:
Artem Bilan
2017-06-21 12:12:49 -04:00
parent a834ce9ffb
commit d766d233f5
3 changed files with 20 additions and 17 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

@@ -60,7 +60,8 @@ public class FluxMessageChannel extends AbstractMessageChannel
@Override
protected boolean doSend(Message<?> message, long timeout) {
Assert.state(subscribers.size() > 0, () -> "The [" + this + "] doesn't have subscribers to accept messages");
Assert.state(this.subscribers.size() > 0,
() -> "The [" + this + "] doesn't have subscribers to accept messages");
this.sink.next(message);
return true;
}

30
spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java
View File

@@ -88,7 +88,14 @@ public class ReactiveStreamsConsumerTests {
reactiveConsumer.stop();
testChannel.send(testMessage);
try {
testChannel.send(testMessage);
}
catch (Exception e) {
assertThat(e, instanceOf(MessageDeliveryException.class));
assertThat(e.getCause(), instanceOf(IllegalStateException.class));
assertThat(e.getMessage(), containsString("doesn't have subscribers to accept messages"));
}
reactiveConsumer.start();
@@ -246,7 +253,14 @@ public class ReactiveStreamsConsumerTests {
endpointFactoryBean.stop();
testChannel.send(testMessage);
try {
testChannel.send(testMessage);
}
catch (Exception e) {
assertThat(e, instanceOf(MessageDeliveryException.class));
assertThat(e.getCause(), instanceOf(IllegalStateException.class));
assertThat(e.getMessage(), containsString("doesn't have subscribers to accept messages"));
}
endpointFactoryBean.start();
@@ -260,16 +274,4 @@ public class ReactiveStreamsConsumerTests {
assertThat(result, Matchers.<Message<?>>contains(testMessage, testMessage2, testMessage2));
}
@Test
public void testFluxMessageChannelSendWithoutSubscription() {
try {
new FluxMessageChannel().send(new GenericMessage<>("foo"));
}
catch (Exception e) {
assertThat(e, instanceOf(MessageDeliveryException.class));
assertThat(e.getCause(), instanceOf(IllegalStateException.class));
assertThat(e.getMessage(), containsString("doesn't have subscribers to accept messages"));
}
}
}

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

@@ -47,8 +47,8 @@ public final class RedisAvailableRule implements MethodRule {
redisStandaloneConfiguration.setPort(REDIS_PORT);
JedisClientConfiguration clientConfiguration = JedisClientConfiguration.builder()
.connectTimeout(Duration.ofSeconds(10))
.readTimeout(Duration.ofSeconds(10))
.connectTimeout(Duration.ofSeconds(20))
.readTimeout(Duration.ofSeconds(20))
.build();
connectionFactory = new JedisConnectionFactory(redisStandaloneConfiguration, clientConfiguration);