diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java index 7297b48d1..44456ac31 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java @@ -424,6 +424,10 @@ public class LettuceConnection extends AbstractRedisConnection { } } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.AbstractRedisConnection#close() + */ @Override public void close() throws DataAccessException { @@ -453,26 +457,48 @@ public class LettuceConnection extends AbstractRedisConnection { this.dbIndex = defaultDbIndex; } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisConnection#isClosed() + */ @Override public boolean isClosed() { return isClosed && !isSubscribed(); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisConnection#getNativeConnection() + */ @Override public RedisClusterAsyncCommands getNativeConnection() { - return (subscription != null ? subscription.pubsub.async() : getAsyncConnection()); + + LettuceSubscription subscription = this.subscription; + return (subscription != null ? subscription.getNativeConnection().async() : getAsyncConnection()); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisConnection#isQueueing() + */ @Override public boolean isQueueing() { return isMulti; } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisConnection#isPipelined() + */ @Override public boolean isPipelined() { return isPipelined; } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisConnection#openPipeline() + */ @Override public void openPipeline() { if (!isPipelined) { @@ -481,6 +507,10 @@ public class LettuceConnection extends AbstractRedisConnection { } } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisConnection#closePipeline() + */ @Override public List closePipeline() { @@ -547,7 +577,6 @@ public class LettuceConnection extends AbstractRedisConnection { * @see org.springframework.data.redis.connection.RedisServerCommands#shutdown(org.springframework.data.redis.connection.RedisServerCommands.ShutdownOption) */ @Override - public byte[] echo(byte[] message) { try { if (isPipelined()) { @@ -714,43 +743,62 @@ public class LettuceConnection extends AbstractRedisConnection { // Pub/Sub functionality // + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisPubSubCommands#publish(byte[], byte[]) + */ @Override public Long publish(byte[] channel, byte[] message) { + try { + if (isPipelined()) { pipeline(newLettuceResult(getAsyncConnection().publish(channel, message))); return null; } + if (isQueueing()) { transaction(newLettuceResult(getAsyncConnection().publish(channel, message))); return null; } + return getConnection().publish(channel, message); } catch (Exception ex) { throw convertLettuceAccessException(ex); } } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisPubSubCommands#getSubscription() + */ @Override public Subscription getSubscription() { return subscription; } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisPubSubCommands#isSubscribed() + */ @Override public boolean isSubscribed() { return (subscription != null && subscription.isAlive()); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisPubSubCommands#pSubscribe(org.springframework.data.redis.connection.MessageListener, byte[][]) + */ @Override public void pSubscribe(MessageListener listener, byte[]... patterns) { + checkSubscription(); - if (isQueueing()) { - throw new UnsupportedOperationException(); - } - if (isPipelined()) { - throw new UnsupportedOperationException(); + if (isQueueing() || isPipelined()) { + throw new UnsupportedOperationException("Transaction/Pipelining is not supported for Pub/Sub subscriptions!"); } + try { subscription = new LettuceSubscription(listener, switchToPubSub(), connectionProvider); subscription.pSubscribe(patterns); @@ -759,17 +807,22 @@ public class LettuceConnection extends AbstractRedisConnection { } } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisPubSubCommands#subscribe(org.springframework.data.redis.connection.MessageListener, byte[][]) + */ @Override public void subscribe(MessageListener listener, byte[]... channels) { + checkSubscription(); - if (isPipelined()) { - throw new UnsupportedOperationException(); + if (isQueueing() || isPipelined()) { + throw new UnsupportedOperationException("Transaction/Pipelining is not supported for Pub/Sub subscriptions!"); } + try { subscription = new LettuceSubscription(listener, switchToPubSub(), connectionProvider); subscription.subscribe(channels); - } catch (Exception ex) { throw convertLettuceAccessException(ex); } diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceMessageListener.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceMessageListener.java index 84bd64f9e..7cee57434 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceMessageListener.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceMessageListener.java @@ -13,7 +13,6 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - package org.springframework.data.redis.connection.lettuce; import io.lettuce.core.pubsub.RedisPubSubListener; @@ -32,23 +31,47 @@ class LettuceMessageListener implements RedisPubSubListener { private final MessageListener listener; LettuceMessageListener(MessageListener listener) { - Assert.notNull(listener, "message listener is required"); + Assert.notNull(listener, "MessageListener must not be null!"); this.listener = listener; } + /* + * (non-Javadoc) + * @see io.lettuce.core.pubsub.RedisPubSubListener#message(java.lang.Object, java.lang.Object) + */ public void message(byte[] channel, byte[] message) { listener.onMessage(new DefaultMessage(channel, message), null); } + /* + * (non-Javadoc) + * @see io.lettuce.core.pubsub.RedisPubSubListener#message(java.lang.Object, java.lang.Object, java.lang.Object) + */ public void message(byte[] pattern, byte[] channel, byte[] message) { listener.onMessage(new DefaultMessage(channel, message), pattern); } + /* + * (non-Javadoc) + * @see io.lettuce.core.pubsub.RedisPubSubListener#subscribed(java.lang.Object, long) + */ public void subscribed(byte[] channel, long count) {} + /* + * (non-Javadoc) + * @see io.lettuce.core.pubsub.RedisPubSubListener#psubscribed(java.lang.Object, long) + */ public void psubscribed(byte[] pattern, long count) {} + /* + * (non-Javadoc) + * @see io.lettuce.core.pubsub.RedisPubSubListener#unsubscribed(java.lang.Object, long) + */ public void unsubscribed(byte[] channel, long count) {} + /* + * (non-Javadoc) + * @see io.lettuce.core.pubsub.RedisPubSubListener#punsubscribed(java.lang.Object, long) + */ public void punsubscribed(byte[] pattern, long count) {} } diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettucePoolingConnectionProvider.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettucePoolingConnectionProvider.java index 052c2d747..f60b8dd1a 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettucePoolingConnectionProvider.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettucePoolingConnectionProvider.java @@ -99,7 +99,7 @@ class LettucePoolingConnectionProvider implements LettuceConnectionProvider, Red } throw new IllegalStateException( - String.format("Undyerlying connection provider %s does not implement RedisClientProvider!", + String.format("Underlying connection provider %s does not implement RedisClientProvider!", connectionProvider.getClass().getName())); } diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceSubscription.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceSubscription.java index 36a5bb244..f6cb2239a 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceSubscription.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceSubscription.java @@ -13,10 +13,10 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - package org.springframework.data.redis.connection.lettuce; import io.lettuce.core.pubsub.StatefulRedisPubSubConnection; +import io.lettuce.core.pubsub.api.sync.RedisPubSubCommands; import org.springframework.data.redis.connection.MessageListener; import org.springframework.data.redis.connection.util.AbstractSubscription; @@ -29,46 +29,78 @@ import org.springframework.data.redis.connection.util.AbstractSubscription; */ class LettuceSubscription extends AbstractSubscription { - final StatefulRedisPubSubConnection pubsub; - private LettuceMessageListener listener; + private final StatefulRedisPubSubConnection connection; + private final LettuceMessageListener listener; private final LettuceConnectionProvider connectionProvider; + private final RedisPubSubCommands pubsub; + + LettuceSubscription(MessageListener listener, StatefulRedisPubSubConnection pubsubConnection, + LettuceConnectionProvider connectionProvider) { - LettuceSubscription(MessageListener listener, StatefulRedisPubSubConnection pubsubConnection, LettuceConnectionProvider connectionProvider) { super(listener); - this.pubsub = pubsubConnection; + + this.connection = pubsubConnection; this.listener = new LettuceMessageListener(listener); this.connectionProvider = connectionProvider; + this.pubsub = connection.sync(); - pubsub.addListener(this.listener); + this.connection.addListener(this.listener); } + protected StatefulRedisPubSubConnection getNativeConnection() { + return connection; + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.util.AbstractSubscription#doClose() + */ protected void doClose() { + if (!getChannels().isEmpty()) { - pubsub.sync().unsubscribe(new byte[0]); + pubsub.unsubscribe(new byte[0]); } + if (!getPatterns().isEmpty()) { - pubsub.sync().punsubscribe(new byte[0]); + pubsub.punsubscribe(new byte[0]); } - pubsub.removeListener(this.listener); - - connectionProvider.release(pubsub); + + connection.removeListener(this.listener); + connectionProvider.release(connection); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.util.AbstractSubscription#doPsubscribe(byte[][]) + */ protected void doPsubscribe(byte[]... patterns) { - pubsub.sync().psubscribe(patterns); + pubsub.psubscribe(patterns); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.util.AbstractSubscription#doPUnsubscribe(boolean, byte[][]) + */ protected void doPUnsubscribe(boolean all, byte[]... patterns) { - // lettuce doesn't automatically subscribe from all channels - pubsub.sync().punsubscribe(patterns); + // lettuce doesn't automatically unsubscribe from all patterns + pubsub.punsubscribe(patterns); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.util.AbstractSubscription#doSubscribe(byte[][]) + */ protected void doSubscribe(byte[]... channels) { - pubsub.sync().subscribe(channels); + pubsub.subscribe(channels); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.util.AbstractSubscription#doUnsubscribe(boolean, byte[][]) + */ protected void doUnsubscribe(boolean all, byte[]... channels) { - // lettuce doesn't automatically subscribe from all patterns - pubsub.sync().unsubscribe(channels); + // lettuce doesn't automatically unsubscribe from all channels + pubsub.unsubscribe(channels); } + }