diff --git a/src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java b/src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java index b4b13849c..99ada10f6 100644 --- a/src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java +++ b/src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java @@ -44,6 +44,7 @@ import org.springframework.beans.factory.InitializingBean; import org.springframework.context.SmartLifecycle; import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.core.task.TaskExecutor; +import org.springframework.dao.DataAccessException; import org.springframework.data.redis.RedisConnectionFailureException; import org.springframework.data.redis.connection.ConnectionUtils; import org.springframework.data.redis.connection.Message; @@ -53,6 +54,7 @@ import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.connection.Subscription; import org.springframework.data.redis.connection.SubscriptionListener; import org.springframework.data.redis.connection.util.ByteArrayWrapper; +import org.springframework.data.redis.listener.adapter.RedisListenerExecutionFailedException; import org.springframework.data.redis.serializer.RedisSerializer; import org.springframework.data.redis.serializer.StringRedisSerializer; import org.springframework.lang.Nullable; @@ -181,7 +183,6 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab subscriptionExecutor = taskExecutor; } - this.subscriber = createSubscriber(connectionFactory, this.subscriptionExecutor); afterPropertiesSet = true; @@ -269,6 +270,11 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab } catch (InterruptedException e) { Thread.currentThread().interrupt(); } catch (ExecutionException e) { + + if (e.getCause() instanceof DataAccessException) { + throw new RedisListenerExecutionFailedException(e.getMessage(), e.getCause()); + } + throw new CompletionException(e.getCause()); } catch (TimeoutException e) { throw new IllegalStateException("Subscription registration timeout exceeded.", e); @@ -670,7 +676,16 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab getRequiredSubscriber().subscribeChannel(channels.toArray(new byte[channels.size()][])); getRequiredSubscriber().subscribePattern(patterns.toArray(new byte[patterns.size()][])); - future.join(); + try { + future.join(); + } catch (CompletionException e) { + + if (e.getCause() instanceof DataAccessException) { + throw new RedisListenerExecutionFailedException(e.getMessage(), e.getCause()); + } + + throw e; + } } } } @@ -1166,23 +1181,25 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab synchronized (localMonitor) { - RedisConnection connection = connectionFactory.getConnection(); - this.connection = connection; - - if (connection.isSubscribed()) { - - CompletableFuture failure = new CompletableFuture<>(); - failure.completeExceptionally( - new IllegalStateException("Retrieved connection is already subscribed; aborting listening")); - return failure; - } - CompletableFuture initFuture = new CompletableFuture<>(); - try { - eventuallyPerformSubscription(connection, backOffExecution, initFuture, patterns, channels); - } catch (Throwable t) { - handleSubscriptionException(initFuture, backOffExecution, t); + RedisConnection connection = connectionFactory.getConnection(); + this.connection = connection; + + if (connection.isSubscribed()) { + + initFuture.completeExceptionally( + new IllegalStateException("Retrieved connection is already subscribed; aborting listening")); + return initFuture; + } + + try { + eventuallyPerformSubscription(connection, backOffExecution, initFuture, patterns, channels); + } catch (Throwable t) { + handleSubscriptionException(initFuture, backOffExecution, t); + } + } catch (RuntimeException e) { + initFuture.completeExceptionally(e); } return initFuture; diff --git a/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerUnitTests.java b/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerUnitTests.java index 09031c276..8d77dd613 100644 --- a/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerUnitTests.java +++ b/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerUnitTests.java @@ -25,12 +25,14 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.springframework.core.task.SyncTaskExecutor; +import org.springframework.data.redis.RedisConnectionFailureException; import org.springframework.data.redis.connection.RedisConnection; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.connection.Subscription; import org.springframework.data.redis.connection.SubscriptionListener; import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; import org.springframework.data.redis.listener.adapter.MessageListenerAdapter; +import org.springframework.data.redis.listener.adapter.RedisListenerExecutionFailedException; /** * Unit tests for {@link RedisMessageListenerContainer}. @@ -106,6 +108,46 @@ class RedisMessageListenerContainerUnitTests { verify(connectionMock).close(); } + @Test // GH-2335 + void containerStartShouldReportFailureOnRedisUnavailability() { + + when(connectionFactoryMock.getConnection()).thenThrow(new RedisConnectionFailureException("Booh!")); + + doAnswer(it -> { + + Runnable r = it.getArgument(0); + r.run(); + return null; + }).when(executorMock).execute(any()); + + container.addMessageListener(adapter, new ChannelTopic("a")); + assertThatExceptionOfType(RedisListenerExecutionFailedException.class).isThrownBy(() -> container.start()); + + assertThat(container.isRunning()).isTrue(); + assertThat(container.isListening()).isFalse(); + } + + @Test // GH-2335 + void containerListenShouldReportFailureOnRedisUnavailability() { + + when(connectionFactoryMock.getConnection()).thenThrow(new RedisConnectionFailureException("Booh!")); + + doAnswer(it -> { + + Runnable r = it.getArgument(0); + r.run(); + return null; + }).when(executorMock).execute(any()); + + container.start(); + + assertThatExceptionOfType(RedisListenerExecutionFailedException.class) + .isThrownBy(() -> container.addMessageListener(adapter, new ChannelTopic("a"))); + + assertThat(container.isRunning()).isTrue(); + assertThat(container.isListening()).isFalse(); + } + @Test // GH-964 void failsOnDuplicateInit() { assertThatExceptionOfType(IllegalStateException.class).isThrownBy(() -> container.afterPropertiesSet());