diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/JedisSubscription.java b/src/main/java/org/springframework/data/redis/connection/jedis/JedisSubscription.java index 7d673a133..d6fb501f7 100644 --- a/src/main/java/org/springframework/data/redis/connection/jedis/JedisSubscription.java +++ b/src/main/java/org/springframework/data/redis/connection/jedis/JedisSubscription.java @@ -36,6 +36,11 @@ class JedisSubscription extends AbstractSubscription { this.jedisPubSub = jedisPubSub; } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.util.AbstractSubscription#doClose() + */ + @Override protected void doClose() { if (!getChannels().isEmpty()) { jedisPubSub.unsubscribe(); @@ -45,10 +50,20 @@ class JedisSubscription extends AbstractSubscription { } } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.util.AbstractSubscription#doPsubscribe(byte[][]) + */ + @Override protected void doPsubscribe(byte[]... patterns) { jedisPubSub.psubscribe(patterns); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.util.AbstractSubscription#doPUnsubscribe(boolean, byte[][]) + */ + @Override protected void doPUnsubscribe(boolean all, byte[]... patterns) { if (all) { jedisPubSub.punsubscribe(); @@ -57,10 +72,20 @@ class JedisSubscription extends AbstractSubscription { } } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.util.AbstractSubscription#doSubscribe(byte[][]) + */ + @Override protected void doSubscribe(byte[]... channels) { jedisPubSub.subscribe(channels); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.util.AbstractSubscription#doUnsubscribe(boolean, byte[][]) + */ + @Override protected void doUnsubscribe(boolean all, byte[]... channels) { if (all) { jedisPubSub.unsubscribe(); 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 73a69ee5c..6f8543360 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 @@ -67,6 +67,7 @@ public class LettuceSubscription extends AbstractSubscription { * (non-Javadoc) * @see org.springframework.data.redis.connection.util.AbstractSubscription#doClose() */ + @Override protected void doClose() { if (!getChannels().isEmpty()) { @@ -85,6 +86,7 @@ public class LettuceSubscription extends AbstractSubscription { * (non-Javadoc) * @see org.springframework.data.redis.connection.util.AbstractSubscription#doPsubscribe(byte[][]) */ + @Override protected void doPsubscribe(byte[]... patterns) { pubsub.psubscribe(patterns); } @@ -93,6 +95,7 @@ public class LettuceSubscription extends AbstractSubscription { * (non-Javadoc) * @see org.springframework.data.redis.connection.util.AbstractSubscription#doPUnsubscribe(boolean, byte[][]) */ + @Override protected void doPUnsubscribe(boolean all, byte[]... patterns) { if (all) { @@ -106,6 +109,7 @@ public class LettuceSubscription extends AbstractSubscription { * (non-Javadoc) * @see org.springframework.data.redis.connection.util.AbstractSubscription#doSubscribe(byte[][]) */ + @Override protected void doSubscribe(byte[]... channels) { pubsub.subscribe(channels); } @@ -114,6 +118,7 @@ public class LettuceSubscription extends AbstractSubscription { * (non-Javadoc) * @see org.springframework.data.redis.connection.util.AbstractSubscription#doUnsubscribe(boolean, byte[][]) */ + @Override protected void doUnsubscribe(boolean all, byte[]... channels) { if (all) { @@ -122,5 +127,4 @@ public class LettuceSubscription extends AbstractSubscription { pubsub.unsubscribe(channels); } } - } diff --git a/src/main/java/org/springframework/data/redis/connection/util/AbstractSubscription.java b/src/main/java/org/springframework/data/redis/connection/util/AbstractSubscription.java index 924a83ba6..95ee33f54 100644 --- a/src/main/java/org/springframework/data/redis/connection/util/AbstractSubscription.java +++ b/src/main/java/org/springframework/data/redis/connection/util/AbstractSubscription.java @@ -111,22 +111,42 @@ public abstract class AbstractSubscription implements Subscription { */ protected abstract void doClose(); + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.Subscription#getListener() + */ + @Override public MessageListener getListener() { return listener; } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.Subscription#getChannels() + */ + @Override public Collection getChannels() { synchronized (channels) { return clone(channels); } } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.Subscription#getPatterns() + */ + @Override public Collection getPatterns() { synchronized (patterns) { return clone(patterns); } } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.Subscription#pSubscribe(byte[][]) + */ + @Override public void pSubscribe(byte[]... patterns) { checkPulse(); @@ -139,10 +159,20 @@ public abstract class AbstractSubscription implements Subscription { doPsubscribe(patterns); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.Subscription#pUnsubscribe() + */ + @Override public void pUnsubscribe() { pUnsubscribe((byte[][]) null); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.Subscription#subscribe(byte[][]) + */ + @Override public void subscribe(byte[]... channels) { checkPulse(); @@ -155,10 +185,20 @@ public abstract class AbstractSubscription implements Subscription { doSubscribe(channels); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.Subscription#unsubscribe() + */ + @Override public void unsubscribe() { unsubscribe((byte[][]) null); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.Subscription#pUnsubscribe(byte[][]) + */ + @Override public void pUnsubscribe(@Nullable byte[]... patts) { if (!isAlive()) { return; @@ -186,6 +226,11 @@ public abstract class AbstractSubscription implements Subscription { closeIfUnsubscribed(); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.Subscription#unsubscribe(byte[][]) + */ + @Override public void unsubscribe(@Nullable byte[]... chans) { if (!isAlive()) { return; @@ -213,6 +258,11 @@ public abstract class AbstractSubscription implements Subscription { closeIfUnsubscribed(); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.Subscription#isAlive() + */ + @Override public boolean isAlive() { return alive.get(); } diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSubscriptionTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSubscriptionTests.java index 5b192c864..5018b5fc5 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSubscriptionTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSubscriptionTests.java @@ -25,7 +25,6 @@ import java.util.Collection; import org.junit.Before; import org.junit.Test; -import org.mockito.Mockito; import org.springframework.data.redis.connection.MessageListener; import org.springframework.data.redis.connection.RedisInvalidSubscriptionException; @@ -41,9 +40,7 @@ public class LettuceSubscriptionTests { private LettuceSubscription subscription; - StatefulRedisPubSubConnection pubsub; - - private MessageListener listener; + private StatefulRedisPubSubConnection pubsub; private RedisPubSubCommands asyncCommands; @@ -53,126 +50,159 @@ public class LettuceSubscriptionTests { @Before public void setUp() { - pubsub = Mockito.mock(StatefulRedisPubSubConnection.class); - listener = Mockito.mock(MessageListener.class); - asyncCommands = Mockito.mock(RedisPubSubCommands.class); - connectionProvider = Mockito.mock(LettuceConnectionProvider.class); + pubsub = mock(StatefulRedisPubSubConnection.class); + asyncCommands = mock(RedisPubSubCommands.class); + connectionProvider = mock(LettuceConnectionProvider.class); - Mockito.when(pubsub.sync()).thenReturn(asyncCommands); - subscription = new LettuceSubscription(listener, pubsub, connectionProvider); + when(pubsub.sync()).thenReturn(asyncCommands); + subscription = new LettuceSubscription(mock(MessageListener.class), pubsub, connectionProvider); } @Test public void testUnsubscribeAllAndClose() { + subscription.subscribe(new byte[][] { "a".getBytes() }); subscription.unsubscribe(); + verify(asyncCommands).unsubscribe(); verify(asyncCommands, never()).punsubscribe(); verify(connectionProvider).release(pubsub); verify(pubsub).removeListener(any(LettuceMessageListener.class)); + assertThat(subscription.isAlive()).isFalse(); - assertThat(subscription.getChannels().isEmpty()).isTrue(); - assertThat(subscription.getPatterns().isEmpty()).isTrue(); + assertThat(subscription.getChannels()).isEmpty(); + assertThat(subscription.getPatterns()).isEmpty(); } @Test public void testUnsubscribeAllChannelsWithPatterns() { + subscription.subscribe(new byte[][] { "a".getBytes() }); subscription.pSubscribe(new byte[][] { "s*".getBytes() }); subscription.unsubscribe(); + verify(asyncCommands).unsubscribe(); verify(asyncCommands, never()).punsubscribe(); + assertThat(subscription.isAlive()).isTrue(); - assertThat(subscription.getChannels().isEmpty()).isTrue(); + assertThat(subscription.getChannels()).isEmpty(); + Collection patterns = subscription.getPatterns(); - assertThat(patterns.size()).isEqualTo(1); + assertThat(patterns).hasSize(1); assertThat(patterns.iterator().next()).isEqualTo("s*".getBytes()); } @Test public void testUnsubscribeChannelAndClose() { + byte[][] channel = new byte[][] { "a".getBytes() }; + subscription.subscribe(channel); subscription.unsubscribe(channel); + verify(asyncCommands).unsubscribe(channel); verify(asyncCommands, never()).unsubscribe(); verify(asyncCommands, never()).punsubscribe(); verify(connectionProvider).release(pubsub); verify(pubsub).removeListener(any(LettuceMessageListener.class)); + assertThat(subscription.isAlive()).isFalse(); - assertThat(subscription.getChannels().isEmpty()).isTrue(); - assertThat(subscription.getPatterns().isEmpty()).isTrue(); + assertThat(subscription.getChannels()).isEmpty(); + assertThat(subscription.getPatterns()).isEmpty(); } @Test public void testUnsubscribeChannelSomeLeft() { + byte[][] channels = new byte[][] { "a".getBytes(), "b".getBytes() }; + subscription.subscribe(channels); subscription.unsubscribe(new byte[][] { "a".getBytes() }); + verify(asyncCommands).unsubscribe(new byte[][] { "a".getBytes() }); verify(asyncCommands, never()).unsubscribe(); verify(asyncCommands, never()).punsubscribe(); + assertThat(subscription.isAlive()).isTrue(); + Collection subChannels = subscription.getChannels(); - assertThat(subChannels.size()).isEqualTo(1); + assertThat(subChannels).hasSize(1); assertThat(subChannels.iterator().next()).isEqualTo("b".getBytes()); - assertThat(subscription.getPatterns().isEmpty()).isTrue(); + assertThat(subscription.getPatterns()).isEmpty(); } @Test public void testUnsubscribeChannelWithPatterns() { + byte[][] channel = new byte[][] { "a".getBytes() }; + subscription.subscribe(channel); subscription.pSubscribe(new byte[][] { "s*".getBytes() }); subscription.unsubscribe(channel); + verify(asyncCommands).unsubscribe(channel); verify(asyncCommands, never()).unsubscribe(); verify(asyncCommands, never()).punsubscribe(); + assertThat(subscription.isAlive()).isTrue(); - assertThat(subscription.getChannels().isEmpty()).isTrue(); + assertThat(subscription.getChannels()).isEmpty(); + Collection patterns = subscription.getPatterns(); - assertThat(patterns.size()).isEqualTo(1); + assertThat(patterns).hasSize(1); assertThat(patterns.iterator().next()).isEqualTo("s*".getBytes()); } @Test public void testUnsubscribeChannelWithPatternsSomeLeft() { + byte[][] channel = new byte[][] { "a".getBytes() }; - subscription.subscribe(new byte[][] { "a".getBytes(), "b".getBytes() }); + + subscription.subscribe("a".getBytes(), "b".getBytes()); subscription.pSubscribe(new byte[][] { "s*".getBytes() }); subscription.unsubscribe(channel); + verify(asyncCommands).unsubscribe(channel); verify(asyncCommands, never()).unsubscribe(); verify(asyncCommands, never()).punsubscribe(); assertThat(subscription.isAlive()).isTrue(); + Collection channels = subscription.getChannels(); - assertThat(channels.size()).isEqualTo(1); + assertThat(channels).hasSize(1); assertThat(channels.iterator().next()).isEqualTo("b".getBytes()); + Collection patterns = subscription.getPatterns(); - assertThat(patterns.size()).isEqualTo(1); + assertThat(patterns).hasSize(1); assertThat(patterns.iterator().next()).isEqualTo("s*".getBytes()); } @Test public void testUnsubscribeAllNoChannels() { + subscription.pSubscribe(new byte[][] { "s*".getBytes() }); subscription.unsubscribe(); + verify(asyncCommands, never()).unsubscribe(); verify(asyncCommands, never()).punsubscribe(); + assertThat(subscription.isAlive()).isTrue(); - assertThat(subscription.getChannels().isEmpty()).isTrue(); + assertThat(subscription.getChannels()).isEmpty(); + Collection patterns = subscription.getPatterns(); - assertThat(patterns.size()).isEqualTo(1); + assertThat(patterns).hasSize(1); assertThat(patterns.iterator().next()).isEqualTo("s*".getBytes()); } @Test public void testUnsubscribeNotAlive() { + subscription.subscribe(new byte[][] { "a".getBytes() }); subscription.unsubscribe(); + verify(connectionProvider).release(pubsub); verify(pubsub).removeListener(any(LettuceMessageListener.class)); + assertThat(subscription.isAlive()).isFalse(); + subscription.unsubscribe(); verify(asyncCommands).unsubscribe(); verify(asyncCommands, never()).punsubscribe(); @@ -180,122 +210,157 @@ public class LettuceSubscriptionTests { @Test(expected = RedisInvalidSubscriptionException.class) public void testSubscribeNotAlive() { + subscription.subscribe(new byte[][] { "a".getBytes() }); subscription.unsubscribe(); + assertThat(subscription.isAlive()).isFalse(); subscription.subscribe(new byte[][] { "s".getBytes() }); } @Test public void testPUnsubscribeAllAndClose() { + subscription.pSubscribe(new byte[][] { "a*".getBytes() }); subscription.pUnsubscribe(); + verify(asyncCommands, never()).unsubscribe(); verify(asyncCommands).punsubscribe(); - assertThat(subscription.isAlive()).isFalse(); verify(connectionProvider).release(pubsub); verify(pubsub).removeListener(any(LettuceMessageListener.class)); - assertThat(subscription.getChannels().isEmpty()).isTrue(); - assertThat(subscription.getPatterns().isEmpty()).isTrue(); + + assertThat(subscription.isAlive()).isFalse(); + assertThat(subscription.getChannels()).isEmpty(); + assertThat(subscription.getPatterns()).isEmpty(); } @Test public void testPUnsubscribeAllPatternsWithChannels() { + subscription.subscribe(new byte[][] { "a".getBytes() }); subscription.pSubscribe(new byte[][] { "s*".getBytes() }); subscription.pUnsubscribe(); + verify(asyncCommands, never()).unsubscribe(); verify(asyncCommands).punsubscribe(); + assertThat(subscription.isAlive()).isTrue(); - assertThat(subscription.getPatterns().isEmpty()).isTrue(); + assertThat(subscription.getPatterns()).isEmpty(); + Collection channels = subscription.getChannels(); - assertThat(channels.size()).isEqualTo(1); + assertThat(channels).hasSize(1); assertThat(channels.iterator().next()).isEqualTo("a".getBytes()); } @Test public void testPUnsubscribeAndClose() { + byte[][] pattern = new byte[][] { "a*".getBytes() }; + subscription.pSubscribe(pattern); subscription.pUnsubscribe(pattern); + verify(asyncCommands, never()).unsubscribe(); verify(asyncCommands, never()).punsubscribe(); verify(asyncCommands).punsubscribe(pattern); verify(connectionProvider).release(pubsub); verify(pubsub).removeListener(any(LettuceMessageListener.class)); + assertThat(subscription.isAlive()).isFalse(); - assertThat(subscription.getChannels().isEmpty()).isTrue(); - assertThat(subscription.getPatterns().isEmpty()).isTrue(); + assertThat(subscription.getChannels()).isEmpty(); + assertThat(subscription.getPatterns()).isEmpty(); } @Test public void testPUnsubscribePatternSomeLeft() { + byte[][] patterns = new byte[][] { "a*".getBytes(), "b*".getBytes() }; subscription.pSubscribe(patterns); subscription.pUnsubscribe(new byte[][] { "a*".getBytes() }); + verify(asyncCommands).punsubscribe(new byte[][] { "a*".getBytes() }); verify(asyncCommands, never()).unsubscribe(); verify(asyncCommands, never()).punsubscribe(); + assertThat(subscription.isAlive()).isTrue(); + Collection subPatterns = subscription.getPatterns(); - assertThat(subPatterns.size()).isEqualTo(1); + assertThat(subPatterns).hasSize(1); assertThat(subPatterns.iterator().next()).isEqualTo("b*".getBytes()); - assertThat(subscription.getChannels().isEmpty()).isTrue(); + assertThat(subscription.getChannels()).isEmpty(); } @Test public void testPUnsubscribePatternWithChannels() { + byte[][] pattern = new byte[][] { "s*".getBytes() }; + subscription.subscribe(new byte[][] { "a".getBytes() }); subscription.pSubscribe(pattern); subscription.pUnsubscribe(pattern); + verify(asyncCommands).punsubscribe(pattern); verify(asyncCommands, never()).unsubscribe(); verify(asyncCommands, never()).punsubscribe(); + assertThat(subscription.isAlive()).isTrue(); - assertThat(subscription.getPatterns().isEmpty()).isTrue(); + assertThat(subscription.getPatterns()).isEmpty(); + Collection channels = subscription.getChannels(); - assertThat(channels.size()).isEqualTo(1); + assertThat(channels).hasSize(1); assertThat(channels.iterator().next()).isEqualTo("a".getBytes()); } @Test public void testUnsubscribePatternWithChannelsSomeLeft() { + byte[][] pattern = new byte[][] { "a*".getBytes() }; - subscription.pSubscribe(new byte[][] { "a*".getBytes(), "b*".getBytes() }); + + subscription.pSubscribe("a*".getBytes(), "b*".getBytes()); subscription.subscribe(new byte[][] { "a".getBytes() }); subscription.pUnsubscribe(pattern); + verify(asyncCommands, never()).unsubscribe(); verify(asyncCommands, never()).punsubscribe(); verify(asyncCommands).punsubscribe(pattern); + assertThat(subscription.isAlive()).isTrue(); + Collection channels = subscription.getChannels(); - assertThat(channels.size()).isEqualTo(1); + assertThat(channels).hasSize(1); assertThat(channels.iterator().next()).isEqualTo("a".getBytes()); + Collection patterns = subscription.getPatterns(); - assertThat(patterns.size()).isEqualTo(1); + assertThat(patterns).hasSize(1); assertThat(patterns.iterator().next()).isEqualTo("b*".getBytes()); } @Test public void testPUnsubscribeAllNoPatterns() { + subscription.subscribe(new byte[][] { "s".getBytes() }); subscription.pUnsubscribe(); + verify(asyncCommands, never()).unsubscribe(); verify(asyncCommands, never()).punsubscribe(); assertThat(subscription.isAlive()).isTrue(); - assertThat(subscription.getPatterns().isEmpty()).isTrue(); + assertThat(subscription.getPatterns()).isEmpty(); + Collection channels = subscription.getChannels(); - assertThat(channels.size()).isEqualTo(1); + assertThat(channels).hasSize(1); assertThat(channels.iterator().next()).isEqualTo("s".getBytes()); } @Test public void testPUnsubscribeNotAlive() { + subscription.subscribe(new byte[][] { "a".getBytes() }); subscription.unsubscribe(); + assertThat(subscription.isAlive()).isFalse(); + subscription.pUnsubscribe(); + verify(connectionProvider).release(pubsub); verify(pubsub).removeListener(any(LettuceMessageListener.class)); verify(asyncCommands).unsubscribe(); @@ -304,31 +369,40 @@ public class LettuceSubscriptionTests { @Test(expected = RedisInvalidSubscriptionException.class) public void testPSubscribeNotAlive() { + subscription.subscribe(new byte[][] { "a".getBytes() }); subscription.unsubscribe(); + assertThat(subscription.isAlive()).isFalse(); + subscription.pSubscribe(new byte[][] { "s*".getBytes() }); } @Test public void testDoCloseNotSubscribed() { + subscription.doClose(); + verify(asyncCommands, never()).unsubscribe(); verify(asyncCommands, never()).punsubscribe(); } @Test public void testDoCloseSubscribedChannels() { + subscription.subscribe(new byte[][] { "a".getBytes() }); subscription.doClose(); + verify(asyncCommands).unsubscribe(); verify(asyncCommands, never()).punsubscribe(); } @Test public void testDoCloseSubscribedPatterns() { + subscription.pSubscribe(new byte[][] { "a*".getBytes() }); subscription.doClose(); + verify(asyncCommands, never()).unsubscribe(); verify(asyncCommands).punsubscribe(); }