diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/UncategorizedRedisException.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/RedisSystemException.java similarity index 85% rename from spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/UncategorizedRedisException.java rename to spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/RedisSystemException.java index 664a23403..b72123868 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/UncategorizedRedisException.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/RedisSystemException.java @@ -23,9 +23,9 @@ import org.springframework.data.keyvalue.UncategorizedKeyvalueStoreException; * * @author Costin Leau */ -public class UncategorizedRedisException extends UncategorizedKeyvalueStoreException { +public class RedisSystemException extends UncategorizedKeyvalueStoreException { - public UncategorizedRedisException(String msg, Throwable cause) { + public RedisSystemException(String msg, Throwable cause) { super(msg, cause); } } diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/DefaultStringRedisConnection.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/DefaultStringRedisConnection.java index ef24430de..eb70967f3 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/DefaultStringRedisConnection.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/DefaultStringRedisConnection.java @@ -23,7 +23,7 @@ import java.util.Map; import java.util.Properties; import java.util.Set; -import org.springframework.data.keyvalue.redis.UncategorizedRedisException; +import org.springframework.data.keyvalue.redis.RedisSystemException; import org.springframework.data.keyvalue.redis.serializer.RedisSerializer; import org.springframework.data.keyvalue.redis.serializer.SerializationUtils; import org.springframework.data.keyvalue.redis.serializer.StringRedisSerializer; @@ -88,7 +88,7 @@ public class DefaultStringRedisConnection implements StringRedisConnection { return delegate.bRPopLPush(timeout, srcKey, dstKey); } - public void close() throws UncategorizedRedisException { + public void close() throws RedisSystemException { delegate.close(); } diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/RedisInvalidSubscriptionException.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/RedisInvalidSubscriptionException.java new file mode 100644 index 000000000..485a87bae --- /dev/null +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/RedisInvalidSubscriptionException.java @@ -0,0 +1,45 @@ +/* + * Copyright 2011 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 + * + * http://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.data.keyvalue.redis.connection; + +import org.springframework.dao.InvalidDataAccessResourceUsageException; + +/** + * Exception thrown when subscribing to an expired/dead {@link Subscription}. + * + * @author Costin Leau + */ +public class RedisInvalidSubscriptionException extends InvalidDataAccessResourceUsageException { + + /** + * Constructs a new RedisInvalidSubscriptionException instance. + * + * @param msg + * @param cause + */ + public RedisInvalidSubscriptionException(String msg, Throwable cause) { + super(msg, cause); + } + + /** + * Constructs a new RedisInvalidSubscriptionException instance. + * + * @param msg + */ + public RedisInvalidSubscriptionException(String msg) { + super(msg); + } +} diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/SubscribedRedisConnectionException.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/RedisSubscribedConnectionException.java similarity index 74% rename from spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/SubscribedRedisConnectionException.java rename to spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/RedisSubscribedConnectionException.java index 2a1945e57..bcc93bab7 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/SubscribedRedisConnectionException.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/RedisSubscribedConnectionException.java @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.springframework.data.keyvalue.redis; +package org.springframework.data.keyvalue.redis.connection; import org.springframework.dao.InvalidDataAccessApiUsageException; @@ -24,24 +24,24 @@ import org.springframework.dao.InvalidDataAccessApiUsageException; * @author Costin Leau * @see org.springframework.data.keyvalue.redis.connection.RedisPubSubCommands */ -public class SubscribedRedisConnectionException extends InvalidDataAccessApiUsageException { +public class RedisSubscribedConnectionException extends InvalidDataAccessApiUsageException { /** - * Constructs a new SubscribedRedisConnectionException instance. + * Constructs a new RedisSubscribedConnectionException instance. * * @param msg * @param cause */ - public SubscribedRedisConnectionException(String msg, Throwable cause) { + public RedisSubscribedConnectionException(String msg, Throwable cause) { super(msg, cause); } /** - * Constructs a new SubscribedRedisConnectionException instance. + * Constructs a new RedisSubscribedConnectionException instance. * * @param msg */ - public SubscribedRedisConnectionException(String msg) { + public RedisSubscribedConnectionException(String msg) { super(msg); } } diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/Subscription.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/Subscription.java index 3000820f1..bdad9ae35 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/Subscription.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/Subscription.java @@ -18,7 +18,10 @@ package org.springframework.data.keyvalue.redis.connection; import java.util.Collection; /** - * Subscription for Redis channels. + * Subscription for Redis channels. Just like the underlying {@link RedisConnection}, + * it should not be used by multiple threads. + * + * Note that once a subscription died, it cannot accept any more subscriptions. * * @author Costin Leau */ @@ -29,14 +32,14 @@ public interface Subscription { * * @param channels channel names */ - void subscribe(byte[]... channels); + void subscribe(byte[]... channels) throws RedisInvalidSubscriptionException; /** * Adds the given channel patterns to the current subscription. * * @param patterns channel patterns */ - void pSubscribe(byte[]... patterns); + void pSubscribe(byte[]... patterns) throws RedisInvalidSubscriptionException; /** * Cancels the current subscription for all channels given by name. diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisConnection.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisConnection.java index 3cf0a1c08..41f1c82e5 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisConnection.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisConnection.java @@ -26,11 +26,11 @@ import java.util.Set; import org.springframework.dao.DataAccessException; import org.springframework.data.keyvalue.UncategorizedKeyvalueStoreException; -import org.springframework.data.keyvalue.redis.SubscribedRedisConnectionException; import org.springframework.data.keyvalue.redis.connection.DataType; import org.springframework.data.keyvalue.redis.connection.MessageListener; import org.springframework.data.keyvalue.redis.connection.RedisConnection; import org.springframework.data.keyvalue.redis.connection.SortParameters; +import org.springframework.data.keyvalue.redis.connection.RedisSubscribedConnectionException; import org.springframework.data.keyvalue.redis.connection.Subscription; import org.springframework.util.ReflectionUtils; @@ -2205,7 +2205,7 @@ public class JedisConnection implements RedisConnection { @Override public void pSubscribe(MessageListener listener, byte[]... patterns) { if (isSubscribed()) { - throw new SubscribedRedisConnectionException( + throw new RedisSubscribedConnectionException( "Connection already subscribed; use the connection Subscription to cancel or add new channels"); } @@ -2229,7 +2229,7 @@ public class JedisConnection implements RedisConnection { @Override public void subscribe(MessageListener listener, byte[]... channels) { if (isSubscribed()) { - throw new SubscribedRedisConnectionException( + throw new RedisSubscribedConnectionException( "Connection already subscribed; use the connection Subscription to cancel or add new channels"); } @@ -2252,7 +2252,7 @@ public class JedisConnection implements RedisConnection { private void checkSubscription() { if (isSubscribed()) { - throw new SubscribedRedisConnectionException("Cannot execute command - connection is subscribed"); + throw new RedisSubscribedConnectionException("Cannot execute command - connection is subscribed"); } } } \ No newline at end of file diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisUtils.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisUtils.java index b76e1d08a..06b0011da 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisUtils.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisUtils.java @@ -29,7 +29,7 @@ import java.util.concurrent.TimeoutException; import org.springframework.dao.DataAccessException; import org.springframework.dao.InvalidDataAccessApiUsageException; import org.springframework.data.keyvalue.redis.RedisConnectionFailureException; -import org.springframework.data.keyvalue.redis.UncategorizedRedisException; +import org.springframework.data.keyvalue.redis.RedisSystemException; import org.springframework.data.keyvalue.redis.connection.DefaultTuple; import org.springframework.data.keyvalue.redis.connection.MessageListener; import org.springframework.data.keyvalue.redis.connection.SortParameters; @@ -87,7 +87,7 @@ public abstract class JedisUtils { return convertJedisAccessException((JedisException) ex); } - return new UncategorizedRedisException("Unknown exception", ex); + return new RedisSystemException("Unknown exception", ex); } static DataAccessException convertJedisAccessException(IOException ex) { @@ -198,7 +198,7 @@ public abstract class JedisUtils { try { info.load(stringReader); } catch (Exception ex) { - throw new UncategorizedRedisException("Cannot read Redis info", ex); + throw new RedisSystemException("Cannot read Redis info", ex); } finally { stringReader.close(); } diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jredis/JredisConnection.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jredis/JredisConnection.java index 4339dcdf5..3d28fb2a7 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jredis/JredisConnection.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jredis/JredisConnection.java @@ -31,7 +31,7 @@ import org.jredis.Query.Support; import org.jredis.ri.alphazero.JRedisService; import org.springframework.dao.DataAccessException; import org.springframework.data.keyvalue.UncategorizedKeyvalueStoreException; -import org.springframework.data.keyvalue.redis.UncategorizedRedisException; +import org.springframework.data.keyvalue.redis.RedisSystemException; import org.springframework.data.keyvalue.redis.connection.DataType; import org.springframework.data.keyvalue.redis.connection.MessageListener; import org.springframework.data.keyvalue.redis.connection.RedisConnection; @@ -75,7 +75,7 @@ public class JredisConnection implements RedisConnection { } @Override - public void close() throws UncategorizedRedisException { + public void close() throws RedisSystemException { isClosed = true; // don't actually close the connection diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcConnection.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcConnection.java index c331b00b8..269456cc3 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcConnection.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcConnection.java @@ -30,11 +30,11 @@ import org.idevlab.rjc.ZParams; import org.idevlab.rjc.message.RedisNodeSubscriber; import org.springframework.dao.DataAccessException; import org.springframework.data.keyvalue.UncategorizedKeyvalueStoreException; -import org.springframework.data.keyvalue.redis.SubscribedRedisConnectionException; import org.springframework.data.keyvalue.redis.connection.DataType; import org.springframework.data.keyvalue.redis.connection.MessageListener; import org.springframework.data.keyvalue.redis.connection.RedisConnection; import org.springframework.data.keyvalue.redis.connection.SortParameters; +import org.springframework.data.keyvalue.redis.connection.RedisSubscribedConnectionException; import org.springframework.data.keyvalue.redis.connection.Subscription; /** @@ -54,6 +54,8 @@ public class RjcConnection implements RedisConnection { private volatile RjcSubscription subscription; private volatile RedisNodeSubscriber subscriber; + private final Object pubSubMonitor = new Object(); + public RjcConnection(org.idevlab.rjc.ds.RedisConnection connection, int dbIndex) { SingleDataSource connectionDataSource = new SingleDataSource(connection); session = new SessionFactoryImpl(connectionDataSource).create(); @@ -1995,7 +1997,7 @@ public class RjcConnection implements RedisConnection { @Override public void pSubscribe(MessageListener listener, byte[]... patterns) { if (isSubscribed()) { - throw new SubscribedRedisConnectionException( + throw new RedisSubscribedConnectionException( "Connection already subscribed; use the connection Subscription to cancel or add new channels"); } @@ -2007,9 +2009,12 @@ public class RjcConnection implements RedisConnection { throw new UnsupportedOperationException(); } - subscription = new RjcSubscription(listener, subscriber); + subscription = new RjcSubscription(listener, subscriber, pubSubMonitor); subscription.pSubscribe(patterns); + synchronized (pubSubMonitor) { + pubSubMonitor.wait(); + } } catch (Exception ex) { throw convertRjcAccessException(ex); } @@ -2018,7 +2023,7 @@ public class RjcConnection implements RedisConnection { @Override public void subscribe(MessageListener listener, byte[]... channels) { if (isSubscribed()) { - throw new SubscribedRedisConnectionException( + throw new RedisSubscribedConnectionException( "Connection already subscribed; use the connection Subscription to cancel or add new channels"); } @@ -2030,8 +2035,12 @@ public class RjcConnection implements RedisConnection { throw new UnsupportedOperationException(); } - subscription = new RjcSubscription(listener, subscriber); + subscription = new RjcSubscription(listener, subscriber, pubSubMonitor); subscription.subscribe(channels); + + synchronized (pubSubMonitor) { + pubSubMonitor.wait(); + } } catch (Exception ex) { throw convertRjcAccessException(ex); @@ -2040,7 +2049,7 @@ public class RjcConnection implements RedisConnection { private void checkSubscription() { if (isSubscribed()) { - throw new SubscribedRedisConnectionException("Cannot execute command - connection is subscribed"); + throw new RedisSubscribedConnectionException("Cannot execute command - connection is subscribed"); } } } \ No newline at end of file diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcUtils.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcUtils.java index afbb7cc68..817d47589 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcUtils.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcUtils.java @@ -31,7 +31,7 @@ import org.idevlab.rjc.ZParams; import org.idevlab.rjc.Client.LIST_POSITION; import org.springframework.dao.DataAccessException; import org.springframework.dao.InvalidDataAccessApiUsageException; -import org.springframework.data.keyvalue.redis.UncategorizedRedisException; +import org.springframework.data.keyvalue.redis.RedisSystemException; import org.springframework.data.keyvalue.redis.connection.DataType; import org.springframework.data.keyvalue.redis.connection.DefaultTuple; import org.springframework.data.keyvalue.redis.connection.SortParameters; @@ -59,7 +59,7 @@ public abstract class RjcUtils { return convertRjcAccessException((RedisException) ex); } - return new UncategorizedRedisException("Unknown exception", ex); + return new RedisSystemException("Unknown exception", ex); } public static DataAccessException convertRjcAccessException(RedisException ex) { @@ -166,7 +166,7 @@ public abstract class RjcUtils { try { info.load(stringReader); } catch (Exception ex) { - throw new UncategorizedRedisException("Cannot read Redis info", ex); + throw new RedisSystemException("Cannot read Redis info", ex); } finally { stringReader.close(); } diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/package-info.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/package-info.java new file mode 100644 index 000000000..66a90b8ae --- /dev/null +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/package-info.java @@ -0,0 +1,5 @@ +/** + * Connection package for RJC library. + */ +package org.springframework.data.keyvalue.redis.connection.rjc; + diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/listener/adapter/MessageListenerAdapter.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/listener/adapter/MessageListenerAdapter.java index 6affa0dc3..8def8aa52 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/listener/adapter/MessageListenerAdapter.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/listener/adapter/MessageListenerAdapter.java @@ -284,11 +284,11 @@ public class MessageListenerAdapter implements MessageListener { throw (DataAccessException) targetEx; } else { - throw new ListenerExecutionFailedException("Listener method '" + methodName + "' threw exception", + throw new RedisListenerExecutionFailedException("Listener method '" + methodName + "' threw exception", targetEx); } } catch (Throwable ex) { - throw new ListenerExecutionFailedException("Failed to invoke target method '" + methodName + throw new RedisListenerExecutionFailedException("Failed to invoke target method '" + methodName + "' with arguments " + ObjectUtils.nullSafeToString(arguments), ex); } } diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/listener/adapter/ListenerExecutionFailedException.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/listener/adapter/RedisListenerExecutionFailedException.java similarity index 71% rename from spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/listener/adapter/ListenerExecutionFailedException.java rename to spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/listener/adapter/RedisListenerExecutionFailedException.java index cb47028bf..8f94a7a95 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/listener/adapter/ListenerExecutionFailedException.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/listener/adapter/RedisListenerExecutionFailedException.java @@ -23,24 +23,24 @@ import org.springframework.dao.InvalidDataAccessApiUsageException; * @author Costin Leau * @see MessageListenerAdapter */ -public class ListenerExecutionFailedException extends InvalidDataAccessApiUsageException { +public class RedisListenerExecutionFailedException extends InvalidDataAccessApiUsageException { /** - * Constructs a new ListenerExecutionFailedException instance. + * Constructs a new RedisListenerExecutionFailedException instance. * * @param msg * @param cause */ - public ListenerExecutionFailedException(String msg, Throwable cause) { + public RedisListenerExecutionFailedException(String msg, Throwable cause) { super(msg, cause); } /** - * Constructs a new ListenerExecutionFailedException instance. + * Constructs a new RedisListenerExecutionFailedException instance. * * @param msg */ - public ListenerExecutionFailedException(String msg) { + public RedisListenerExecutionFailedException(String msg) { super(msg); } }