Fix ReactiveStreamsConsumerTests and Checkstyle

Also increase timeouts in the `RedisAvailableRule`

Author: artembilan

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

@@ -60,7 +60,8 @@ public FluxMessageChannel() {
 
 	@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;
 	}

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

@@ -88,7 +88,14 @@ public void testReactiveStreamsConsumerFluxMessageChannel() throws InterruptedEx
 
 		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 void testReactiveStreamsConsumerViaConsumerEndpointFactoryBean() throws E
 
 		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 void testReactiveStreamsConsumerViaConsumerEndpointFactoryBean() throws E
 		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"));
-		}
-	}
-
 }

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

@@ -47,8 +47,8 @@ public Statement apply(final Statement base, final FrameworkMethod method, Objec
 				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);