diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java b/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java index 6edb1c3d0..ec0aef2ae 100644 --- a/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java @@ -2437,10 +2437,4 @@ public class JedisConnection implements RedisConnection { throw convertJedisAccessException(ex); } } - - private void checkSubscription() { - if (isSubscribed()) { - throw new RedisSubscribedConnectionException("Cannot execute command - connection is subscribed"); - } - } } 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 1cb8eea8d..5540d7911 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,8 +36,12 @@ class JedisSubscription extends AbstractSubscription { protected void doClose() { - jedisPubSub.unsubscribe(); - jedisPubSub.punsubscribe(); + if(!getChannels().isEmpty()) { + jedisPubSub.unsubscribe(); + } + if(!getPatterns().isEmpty()) { + jedisPubSub.punsubscribe(); + } } 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 fdda3b5ef..84180b2da 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 @@ -40,8 +40,12 @@ class LettuceSubscription extends AbstractSubscription { } protected void doClose() { - pubsub.unsubscribe(new byte[0]); - pubsub.punsubscribe(new byte[0]); + if(!getChannels().isEmpty()) { + pubsub.unsubscribe(new byte[0]); + } + if(!getPatterns().isEmpty()) { + pubsub.punsubscribe(new byte[0]); + } pubsub.removeListener(this.listener); pubsub.close(); } diff --git a/src/main/java/org/springframework/data/redis/connection/rjc/RjcConnection.java b/src/main/java/org/springframework/data/redis/connection/rjc/RjcConnection.java index 09518c534..6cacac2dc 100644 --- a/src/main/java/org/springframework/data/redis/connection/rjc/RjcConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/rjc/RjcConnection.java @@ -59,11 +59,38 @@ public class RjcConnection implements RedisConnection { private volatile RjcSubscription subscription; private volatile RedisNodeSubscriber subscriber; + /** + * Constructs a new RjcConnection with a {@link RedisNodeSubscriber} that + * re-uses the passed connection. Not recommended for pooled connections, as + * {@link RedisNodeSubscriber} may leave the Reply InputStream in an + * inconsistent state due to extra unsubscribe calls in close() + * + * @param connection + * The connection to use + * @param dbIndex + * The index of the database to use + */ public RjcConnection(org.idevlab.rjc.ds.RedisConnection connection, int dbIndex) { + this(connection, dbIndex, new RedisNodeSubscriber(new SingleDataSource( + new CloseSuppressingRjcConnection(connection)))); + } + + /** + * Constructs a new RjcConnection with a dedicated + * {@link RedisNodeSubscriber} + * + * @param connection + * The connection to use + * @param dbIndex + * The index of the database to use + * @param subscriber + * The connection to use for subscribe calls + */ + public RjcConnection(org.idevlab.rjc.ds.RedisConnection connection, int dbIndex, + RedisNodeSubscriber subscriber) { SingleDataSource connectionDataSource = new SingleDataSource(connection); session = new SessionFactoryImpl(connectionDataSource).create(); - subscriber = new RedisNodeSubscriber(); - subscriber.setDataSource(new SingleDataSource(new CloseSuppressingRjcConnection(connection))); + this.subscriber = subscriber; client = new Client(connection); this.connection = connection; @@ -2153,10 +2180,4 @@ public class RjcConnection implements RedisConnection { throw convertRjcAccessException(ex); } } - - private void checkSubscription() { - if (isSubscribed()) { - throw new RedisSubscribedConnectionException("Cannot execute command - connection is subscribed"); - } - } } \ No newline at end of file diff --git a/src/main/java/org/springframework/data/redis/connection/rjc/RjcConnectionFactory.java b/src/main/java/org/springframework/data/redis/connection/rjc/RjcConnectionFactory.java index 3adc8a8f7..a85ca2df0 100644 --- a/src/main/java/org/springframework/data/redis/connection/rjc/RjcConnectionFactory.java +++ b/src/main/java/org/springframework/data/redis/connection/rjc/RjcConnectionFactory.java @@ -20,6 +20,7 @@ import org.apache.commons.logging.LogFactory; import org.idevlab.rjc.ds.DataSource; import org.idevlab.rjc.ds.PoolableDataSource; import org.idevlab.rjc.ds.SimpleDataSource; +import org.idevlab.rjc.message.RedisNodeSubscriber; import org.idevlab.rjc.protocol.Protocol; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; @@ -45,6 +46,7 @@ public class RjcConnectionFactory implements InitializingBean, DisposableBean, R private boolean usePool = true; private int dbIndex = 0; private DataSource dataSource; + private DataSource subscriptionDataSource; /** @@ -71,6 +73,7 @@ public class RjcConnectionFactory implements InitializingBean, DisposableBean, R else { dataSource = new SimpleDataSource(hostName, port, timeout, password); } + subscriptionDataSource = new SimpleDataSource(hostName, port, timeout, password); } public void destroy() { @@ -86,7 +89,8 @@ public class RjcConnectionFactory implements InitializingBean, DisposableBean, R public RedisConnection getConnection() { - return postProcessConnection(new RjcConnection(dataSource.getConnection(), dbIndex)); + return postProcessConnection(new RjcConnection(dataSource.getConnection(), dbIndex, + new RedisNodeSubscriber(subscriptionDataSource))); } /** diff --git a/src/main/java/org/springframework/data/redis/connection/rjc/RjcSubscription.java b/src/main/java/org/springframework/data/redis/connection/rjc/RjcSubscription.java index 568ba2dc4..b212ac29a 100644 --- a/src/main/java/org/springframework/data/redis/connection/rjc/RjcSubscription.java +++ b/src/main/java/org/springframework/data/redis/connection/rjc/RjcSubscription.java @@ -37,7 +37,9 @@ class RjcSubscription extends AbstractSubscription { protected void doClose() { - subscriber.close(); + if(!getChannels().isEmpty() || !getPatterns().isEmpty()) { + subscriber.close(); + } } @@ -57,6 +59,6 @@ class RjcSubscription extends AbstractSubscription { protected void doUnsubscribe(boolean all, byte[]... channels) { - subscriber.punsubscribe(RjcUtils.decodeMultiple(channels)); + subscriber.unsubscribe(RjcUtils.decodeMultiple(channels)); } } \ No newline at end of file diff --git a/src/main/java/org/springframework/data/redis/connection/srp/SrpSubscription.java b/src/main/java/org/springframework/data/redis/connection/srp/SrpSubscription.java index 300c2348b..dfdbf4ed2 100644 --- a/src/main/java/org/springframework/data/redis/connection/srp/SrpSubscription.java +++ b/src/main/java/org/springframework/data/redis/connection/srp/SrpSubscription.java @@ -40,8 +40,12 @@ class SrpSubscription extends AbstractSubscription { } protected void doClose() { - client.unsubscribe((Object[]) null); - client.punsubscribe((Object[]) null); + if(!getChannels().isEmpty()) { + client.unsubscribe((Object[]) null); + } + if(!getPatterns().isEmpty()) { + client.punsubscribe((Object[]) null); + } client.removeListener(this.listener); } 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 30cd8508a..1df7d8594 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 @@ -163,8 +163,9 @@ public abstract class AbstractSubscription implements Subscription { // shortcut for unsubscribing all patterns if (ObjectUtils.isEmpty(patts)) { if (!this.patterns.isEmpty()) { - patts = getPatterns().toArray(new byte[this.patterns.size()][]); synchronized (this.patterns) { + patts = getPatterns().toArray(new byte[this.patterns.size()][]); + doPUnsubscribe(true, patts); this.patterns.clear(); } } @@ -174,14 +175,13 @@ public abstract class AbstractSubscription implements Subscription { } } else { + doPUnsubscribe(false, patts); synchronized (this.patterns) { remove(this.patterns, patts); } } - if (isWorking()) { - doPUnsubscribe(this.patterns.isEmpty(), patts); - } + closeIfUnsubscribed(); } @@ -193,8 +193,9 @@ public abstract class AbstractSubscription implements Subscription { // shortcut for unsubscribing all channels if (ObjectUtils.isEmpty(chans)) { if (!this.channels.isEmpty()) { - chans = getChannels().toArray(new byte[this.channels.size()][]); synchronized (this.channels) { + chans = getChannels().toArray(new byte[this.channels.size()][]); + doUnsubscribe(true, chans); this.channels.clear(); } } @@ -204,14 +205,13 @@ public abstract class AbstractSubscription implements Subscription { } } else { + doUnsubscribe(false, chans); synchronized (this.channels) { remove(this.channels, chans); } } - if (isWorking()) { - doUnsubscribe(this.channels.isEmpty(), chans); - } + closeIfUnsubscribed(); } @@ -225,12 +225,11 @@ public abstract class AbstractSubscription implements Subscription { } } - private boolean isWorking() { + private void closeIfUnsubscribed() { if (channels.isEmpty() && patterns.isEmpty()) { alive.set(false); doClose(); } - return isAlive(); } diff --git a/src/test/java/org/springframework/data/redis/connection/jedis/JedisSubscriptionTests.java b/src/test/java/org/springframework/data/redis/connection/jedis/JedisSubscriptionTests.java new file mode 100644 index 000000000..6f6d214fe --- /dev/null +++ b/src/test/java/org/springframework/data/redis/connection/jedis/JedisSubscriptionTests.java @@ -0,0 +1,307 @@ +/* + * Copyright 2011-2013 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.redis.connection.jedis; + +import java.util.Collection; + +import org.junit.Before; +import org.junit.Test; +import org.mockito.Mockito; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.never; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertArrayEquals; +import org.springframework.data.redis.connection.MessageListener; +import org.springframework.data.redis.connection.RedisInvalidSubscriptionException; + +import redis.clients.jedis.BinaryJedisPubSub; + +/** + * Unit test of {@link JedisSubscription} + * + * @author Jennifer Hickey + * + */ +public class JedisSubscriptionTests { + + private JedisSubscription subscription; + + private BinaryJedisPubSub jedisPubSub; + + private MessageListener listener; + + @Before + public void setUp() { + jedisPubSub = Mockito.mock(BinaryJedisPubSub.class); + listener = Mockito.mock(MessageListener.class); + subscription = new JedisSubscription(listener, jedisPubSub, null, null); + } + + @Test + public void testUnsubscribeAllAndClose() { + subscription.subscribe(new byte[][] { "a".getBytes() }); + subscription.unsubscribe(); + verify(jedisPubSub, times(1)).unsubscribe(); + verify(jedisPubSub, never()).punsubscribe(); + assertFalse(subscription.isAlive()); + assertTrue(subscription.getChannels().isEmpty()); + assertTrue(subscription.getPatterns().isEmpty()); + } + + @Test + public void testUnsubscribeAllChannelsWithPatterns() { + subscription.subscribe(new byte[][] { "a".getBytes() }); + subscription.pSubscribe(new byte[][] { "s*".getBytes() }); + subscription.unsubscribe(); + verify(jedisPubSub, times(1)).unsubscribe(); + verify(jedisPubSub, never()).punsubscribe(); + assertTrue(subscription.isAlive()); + assertTrue(subscription.getChannels().isEmpty()); + Collection patterns = subscription.getPatterns(); + assertEquals(1, patterns.size()); + assertArrayEquals("s*".getBytes(), patterns.iterator().next()); + } + + @Test + public void testUnsubscribeChannelAndClose() { + byte[][] channel = new byte[][] { "a".getBytes() }; + subscription.subscribe(channel); + subscription.unsubscribe(channel); + verify(jedisPubSub, times(1)).unsubscribe(channel); + verify(jedisPubSub, never()).punsubscribe(); + assertFalse(subscription.isAlive()); + assertTrue(subscription.getChannels().isEmpty()); + assertTrue(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(jedisPubSub, times(1)).unsubscribe(new byte[][] { "a".getBytes() }); + verify(jedisPubSub, never()).punsubscribe(); + assertTrue(subscription.isAlive()); + Collection subChannels = subscription.getChannels(); + assertEquals(1, subChannels.size()); + assertArrayEquals("b".getBytes(), subChannels.iterator().next()); + assertTrue(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(jedisPubSub, times(1)).unsubscribe(channel); + verify(jedisPubSub, never()).punsubscribe(); + assertTrue(subscription.isAlive()); + assertTrue(subscription.getChannels().isEmpty()); + Collection patterns = subscription.getPatterns(); + assertEquals(1, patterns.size()); + assertArrayEquals("s*".getBytes(), patterns.iterator().next()); + } + + @Test + public void testUnsubscribeChannelWithPatternsSomeLeft() { + byte[][] channel = new byte[][] { "a".getBytes() }; + subscription.subscribe(new byte[][] { "a".getBytes(), "b".getBytes() }); + subscription.pSubscribe(new byte[][] { "s*".getBytes() }); + subscription.unsubscribe(channel); + verify(jedisPubSub, times(1)).unsubscribe(channel); + verify(jedisPubSub, never()).punsubscribe(); + assertTrue(subscription.isAlive()); + Collection channels = subscription.getChannels(); + assertEquals(1, channels.size()); + assertArrayEquals("b".getBytes(), channels.iterator().next()); + Collection patterns = subscription.getPatterns(); + assertEquals(1, patterns.size()); + assertArrayEquals("s*".getBytes(), patterns.iterator().next()); + } + + @Test + public void testUnsubscribeAllNoChannels() { + subscription.pSubscribe(new byte[][] { "s*".getBytes() }); + subscription.unsubscribe(); + verify(jedisPubSub, never()).unsubscribe(); + verify(jedisPubSub, never()).punsubscribe(); + assertTrue(subscription.isAlive()); + assertTrue(subscription.getChannels().isEmpty()); + Collection patterns = subscription.getPatterns(); + assertEquals(1, patterns.size()); + assertArrayEquals("s*".getBytes(), patterns.iterator().next()); + } + + @Test + public void testUnsubscribeNotAlive() { + subscription.subscribe(new byte[][] { "a".getBytes() }); + subscription.unsubscribe(); + assertFalse(subscription.isAlive()); + subscription.unsubscribe(); + verify(jedisPubSub, times(1)).unsubscribe(); + verify(jedisPubSub, never()).punsubscribe(); + } + + @Test(expected = RedisInvalidSubscriptionException.class) + public void testSubscribeNotAlive() { + subscription.subscribe(new byte[][] { "a".getBytes() }); + subscription.unsubscribe(); + assertFalse(subscription.isAlive()); + subscription.subscribe(new byte[][] { "s".getBytes() }); + } + + @Test + public void testPUnsubscribeAllAndClose() { + subscription.pSubscribe(new byte[][] { "a*".getBytes() }); + subscription.pUnsubscribe(); + verify(jedisPubSub, never()).unsubscribe(); + verify(jedisPubSub, times(1)).punsubscribe(); + assertFalse(subscription.isAlive()); + assertTrue(subscription.getChannels().isEmpty()); + assertTrue(subscription.getPatterns().isEmpty()); + } + + @Test + public void testPUnsubscribeAllPatternsWithChannels() { + subscription.subscribe(new byte[][] { "a".getBytes() }); + subscription.pSubscribe(new byte[][] { "s*".getBytes() }); + subscription.pUnsubscribe(); + verify(jedisPubSub, never()).unsubscribe(); + verify(jedisPubSub, times(1)).punsubscribe(); + assertTrue(subscription.isAlive()); + assertTrue(subscription.getPatterns().isEmpty()); + Collection channels = subscription.getChannels(); + assertEquals(1, channels.size()); + assertArrayEquals("a".getBytes(), channels.iterator().next()); + } + + @Test + public void testPUnsubscribeAndClose() { + byte[][] pattern = new byte[][] { "a*".getBytes() }; + subscription.pSubscribe(pattern); + subscription.pUnsubscribe(pattern); + verify(jedisPubSub, never()).unsubscribe(); + verify(jedisPubSub, times(1)).punsubscribe(pattern); + assertFalse(subscription.isAlive()); + assertTrue(subscription.getChannels().isEmpty()); + assertTrue(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(jedisPubSub, times(1)).punsubscribe(new byte[][] { "a*".getBytes() }); + verify(jedisPubSub, never()).unsubscribe(); + assertTrue(subscription.isAlive()); + Collection subPatterns = subscription.getPatterns(); + assertEquals(1, subPatterns.size()); + assertArrayEquals("b*".getBytes(), subPatterns.iterator().next()); + assertTrue(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(jedisPubSub, times(1)).punsubscribe(pattern); + verify(jedisPubSub, never()).unsubscribe(); + assertTrue(subscription.isAlive()); + assertTrue(subscription.getPatterns().isEmpty()); + Collection channels = subscription.getChannels(); + assertEquals(1, channels.size()); + assertArrayEquals("a".getBytes(), channels.iterator().next()); + } + + @Test + public void testUnsubscribePatternWithChannelsSomeLeft() { + byte[][] pattern = new byte[][] { "a*".getBytes() }; + subscription.pSubscribe(new byte[][] { "a*".getBytes(), "b*".getBytes() }); + subscription.subscribe(new byte[][] { "a".getBytes() }); + subscription.pUnsubscribe(pattern); + verify(jedisPubSub, never()).unsubscribe(); + verify(jedisPubSub, times(1)).punsubscribe(pattern); + assertTrue(subscription.isAlive()); + Collection channels = subscription.getChannels(); + assertEquals(1, channels.size()); + assertArrayEquals("a".getBytes(), channels.iterator().next()); + Collection patterns = subscription.getPatterns(); + assertEquals(1, patterns.size()); + assertArrayEquals("b*".getBytes(), patterns.iterator().next()); + } + + @Test + public void testPUnsubscribeAllNoPatterns() { + subscription.subscribe(new byte[][] { "s".getBytes() }); + subscription.pUnsubscribe(); + verify(jedisPubSub, never()).unsubscribe(); + verify(jedisPubSub, never()).punsubscribe(); + assertTrue(subscription.isAlive()); + assertTrue(subscription.getPatterns().isEmpty()); + Collection channels = subscription.getChannels(); + assertEquals(1, channels.size()); + assertArrayEquals("s".getBytes(), channels.iterator().next()); + } + + @Test + public void testPUnsubscribeNotAlive() { + subscription.subscribe(new byte[][] { "a".getBytes() }); + subscription.unsubscribe(); + assertFalse(subscription.isAlive()); + subscription.pUnsubscribe(); + verify(jedisPubSub, times(1)).unsubscribe(); + verify(jedisPubSub, never()).punsubscribe(); + } + + @Test(expected = RedisInvalidSubscriptionException.class) + public void testPSubscribeNotAlive() { + subscription.subscribe(new byte[][] { "a".getBytes() }); + subscription.unsubscribe(); + assertFalse(subscription.isAlive()); + subscription.pSubscribe(new byte[][] { "s*".getBytes() }); + } + + @Test + public void testDoCloseNotSubscribed() { + subscription.doClose(); + verify(jedisPubSub, never()).unsubscribe(); + verify(jedisPubSub, never()).punsubscribe(); + } + + @Test + public void testDoCloseSubscribedChannels() { + subscription.subscribe(new byte[][] { "a".getBytes() }); + subscription.doClose(); + verify(jedisPubSub, times(1)).unsubscribe(); + verify(jedisPubSub, never()).punsubscribe(); + } + + @Test + public void testDoCloseSubscribedPatterns() { + subscription.pSubscribe(new byte[][] { "a*".getBytes() }); + subscription.doClose(); + verify(jedisPubSub, never()).unsubscribe(); + verify(jedisPubSub, times(1)).punsubscribe(); + } + +} 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 new file mode 100644 index 000000000..f31b157d9 --- /dev/null +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSubscriptionTests.java @@ -0,0 +1,335 @@ +/* + * Copyright 2011-2013 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.redis.connection.lettuce; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.any; +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; + +import com.lambdaworks.redis.pubsub.RedisPubSubConnection; + +/** + * Unit test of {@link LettuceSubscription} + * + * @author Jennifer Hickey + * + */ +public class LettuceSubscriptionTests { + + private LettuceSubscription subscription; + + RedisPubSubConnection pubsub; + + private MessageListener listener; + + @SuppressWarnings("unchecked") + @Before + public void setUp() { + pubsub = Mockito.mock(RedisPubSubConnection.class); + listener = Mockito.mock(MessageListener.class); + subscription = new LettuceSubscription(listener, pubsub); + } + + @Test + public void testUnsubscribeAllAndClose() { + subscription.subscribe(new byte[][] { "a".getBytes() }); + subscription.unsubscribe(); + verify(pubsub, times(1)).unsubscribe(new byte[][] { "a".getBytes() }); + verify(pubsub, never()).unsubscribe(new byte[0]); + verify(pubsub, never()).punsubscribe(new byte[0]); + verify(pubsub).close(); + verify(pubsub).removeListener(any(LettuceMessageListener.class)); + assertFalse(subscription.isAlive()); + assertTrue(subscription.getChannels().isEmpty()); + assertTrue(subscription.getPatterns().isEmpty()); + } + + @Test + public void testUnsubscribeAllChannelsWithPatterns() { + subscription.subscribe(new byte[][] { "a".getBytes() }); + subscription.pSubscribe(new byte[][] { "s*".getBytes() }); + subscription.unsubscribe(); + verify(pubsub, times(1)).unsubscribe(new byte[][] { "a".getBytes() }); + verify(pubsub, never()).unsubscribe(new byte[0]); + verify(pubsub, never()).punsubscribe(new byte[0]); + assertTrue(subscription.isAlive()); + assertTrue(subscription.getChannels().isEmpty()); + Collection patterns = subscription.getPatterns(); + assertEquals(1, patterns.size()); + assertArrayEquals("s*".getBytes(), patterns.iterator().next()); + } + + @Test + public void testUnsubscribeChannelAndClose() { + byte[][] channel = new byte[][] { "a".getBytes() }; + subscription.subscribe(channel); + subscription.unsubscribe(channel); + verify(pubsub, times(1)).unsubscribe(channel); + verify(pubsub, never()).unsubscribe(new byte[0]); + verify(pubsub, never()).punsubscribe(new byte[0]); + verify(pubsub).close(); + verify(pubsub).removeListener(any(LettuceMessageListener.class)); + assertFalse(subscription.isAlive()); + assertTrue(subscription.getChannels().isEmpty()); + assertTrue(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(pubsub, times(1)).unsubscribe(new byte[][] { "a".getBytes() }); + verify(pubsub, never()).unsubscribe(new byte[0]); + verify(pubsub, never()).punsubscribe(new byte[0]); + assertTrue(subscription.isAlive()); + Collection subChannels = subscription.getChannels(); + assertEquals(1, subChannels.size()); + assertArrayEquals("b".getBytes(), subChannels.iterator().next()); + assertTrue(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(pubsub, times(1)).unsubscribe(channel); + verify(pubsub, never()).unsubscribe(new byte[0]); + verify(pubsub, never()).punsubscribe(new byte[0]); + assertTrue(subscription.isAlive()); + assertTrue(subscription.getChannels().isEmpty()); + Collection patterns = subscription.getPatterns(); + assertEquals(1, patterns.size()); + assertArrayEquals("s*".getBytes(), patterns.iterator().next()); + } + + @Test + public void testUnsubscribeChannelWithPatternsSomeLeft() { + byte[][] channel = new byte[][] { "a".getBytes() }; + subscription.subscribe(new byte[][] { "a".getBytes(), "b".getBytes() }); + subscription.pSubscribe(new byte[][] { "s*".getBytes() }); + subscription.unsubscribe(channel); + verify(pubsub, times(1)).unsubscribe(channel); + verify(pubsub, never()).unsubscribe(new byte[0]); + verify(pubsub, never()).punsubscribe(new byte[0]); + assertTrue(subscription.isAlive()); + Collection channels = subscription.getChannels(); + assertEquals(1, channels.size()); + assertArrayEquals("b".getBytes(), channels.iterator().next()); + Collection patterns = subscription.getPatterns(); + assertEquals(1, patterns.size()); + assertArrayEquals("s*".getBytes(), patterns.iterator().next()); + } + + @Test + public void testUnsubscribeAllNoChannels() { + subscription.pSubscribe(new byte[][] { "s*".getBytes() }); + subscription.unsubscribe(); + verify(pubsub, never()).unsubscribe(new byte[0]); + verify(pubsub, never()).punsubscribe(new byte[0]); + assertTrue(subscription.isAlive()); + assertTrue(subscription.getChannels().isEmpty()); + Collection patterns = subscription.getPatterns(); + assertEquals(1, patterns.size()); + assertArrayEquals("s*".getBytes(), patterns.iterator().next()); + } + + @Test + public void testUnsubscribeNotAlive() { + subscription.subscribe(new byte[][] { "a".getBytes() }); + subscription.unsubscribe(); + verify(pubsub, times(1)).close(); + verify(pubsub, times(1)).removeListener(any(LettuceMessageListener.class)); + assertFalse(subscription.isAlive()); + subscription.unsubscribe(); + verify(pubsub, times(1)).unsubscribe(new byte[][] { "a".getBytes() }); + verify(pubsub, never()).unsubscribe(new byte[0]); + verify(pubsub, never()).punsubscribe(new byte[0]); + } + + @Test(expected = RedisInvalidSubscriptionException.class) + public void testSubscribeNotAlive() { + subscription.subscribe(new byte[][] { "a".getBytes() }); + subscription.unsubscribe(); + assertFalse(subscription.isAlive()); + subscription.subscribe(new byte[][] { "s".getBytes() }); + } + + @Test + public void testPUnsubscribeAllAndClose() { + subscription.pSubscribe(new byte[][] { "a*".getBytes() }); + subscription.pUnsubscribe(); + verify(pubsub, never()).unsubscribe(new byte[0]); + verify(pubsub, never()).punsubscribe(new byte[0]); + verify(pubsub, times(1)).punsubscribe(new byte[][] { "a*".getBytes() }); + assertFalse(subscription.isAlive()); + verify(pubsub).close(); + verify(pubsub).removeListener(any(LettuceMessageListener.class)); + assertTrue(subscription.getChannels().isEmpty()); + assertTrue(subscription.getPatterns().isEmpty()); + } + + @Test + public void testPUnsubscribeAllPatternsWithChannels() { + subscription.subscribe(new byte[][] { "a".getBytes() }); + subscription.pSubscribe(new byte[][] { "s*".getBytes() }); + subscription.pUnsubscribe(); + verify(pubsub, never()).unsubscribe(new byte[0]); + verify(pubsub, never()).punsubscribe(new byte[0]); + verify(pubsub, times(1)).punsubscribe(new byte[][] { "s*".getBytes() }); + assertTrue(subscription.isAlive()); + assertTrue(subscription.getPatterns().isEmpty()); + Collection channels = subscription.getChannels(); + assertEquals(1, channels.size()); + assertArrayEquals("a".getBytes(), channels.iterator().next()); + } + + @Test + public void testPUnsubscribeAndClose() { + byte[][] pattern = new byte[][] { "a*".getBytes() }; + subscription.pSubscribe(pattern); + subscription.pUnsubscribe(pattern); + verify(pubsub, never()).unsubscribe(new byte[0]); + verify(pubsub, never()).punsubscribe(new byte[0]); + verify(pubsub, times(1)).punsubscribe(pattern); + verify(pubsub).close(); + verify(pubsub).removeListener(any(LettuceMessageListener.class)); + assertFalse(subscription.isAlive()); + assertTrue(subscription.getChannels().isEmpty()); + assertTrue(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(pubsub, times(1)).punsubscribe(new byte[][] { "a*".getBytes() }); + verify(pubsub, never()).unsubscribe(new byte[0]); + verify(pubsub, never()).punsubscribe(new byte[0]); + assertTrue(subscription.isAlive()); + Collection subPatterns = subscription.getPatterns(); + assertEquals(1, subPatterns.size()); + assertArrayEquals("b*".getBytes(), subPatterns.iterator().next()); + assertTrue(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(pubsub, times(1)).punsubscribe(pattern); + verify(pubsub, never()).unsubscribe(new byte[0]); + verify(pubsub, never()).punsubscribe(new byte[0]); + assertTrue(subscription.isAlive()); + assertTrue(subscription.getPatterns().isEmpty()); + Collection channels = subscription.getChannels(); + assertEquals(1, channels.size()); + assertArrayEquals("a".getBytes(), channels.iterator().next()); + } + + @Test + public void testUnsubscribePatternWithChannelsSomeLeft() { + byte[][] pattern = new byte[][] { "a*".getBytes() }; + subscription.pSubscribe(new byte[][] { "a*".getBytes(), "b*".getBytes() }); + subscription.subscribe(new byte[][] { "a".getBytes() }); + subscription.pUnsubscribe(pattern); + verify(pubsub, never()).unsubscribe(new byte[0]); + verify(pubsub, never()).punsubscribe(new byte[0]); + verify(pubsub, times(1)).punsubscribe(pattern); + assertTrue(subscription.isAlive()); + Collection channels = subscription.getChannels(); + assertEquals(1, channels.size()); + assertArrayEquals("a".getBytes(), channels.iterator().next()); + Collection patterns = subscription.getPatterns(); + assertEquals(1, patterns.size()); + assertArrayEquals("b*".getBytes(), patterns.iterator().next()); + } + + @Test + public void testPUnsubscribeAllNoPatterns() { + subscription.subscribe(new byte[][] { "s".getBytes() }); + subscription.pUnsubscribe(); + verify(pubsub, never()).unsubscribe(new byte[0]); + verify(pubsub, never()).punsubscribe(new byte[0]); + assertTrue(subscription.isAlive()); + assertTrue(subscription.getPatterns().isEmpty()); + Collection channels = subscription.getChannels(); + assertEquals(1, channels.size()); + assertArrayEquals("s".getBytes(), channels.iterator().next()); + } + + @Test + public void testPUnsubscribeNotAlive() { + subscription.subscribe(new byte[][] { "a".getBytes() }); + subscription.unsubscribe(); + assertFalse(subscription.isAlive()); + subscription.pUnsubscribe(); + verify(pubsub, times(1)).close(); + verify(pubsub, times(1)).removeListener(any(LettuceMessageListener.class)); + verify(pubsub, times(1)).unsubscribe(new byte[][] { "a".getBytes() }); + verify(pubsub, never()).unsubscribe(new byte[0]); + verify(pubsub, never()).punsubscribe(new byte[0]); + } + + @Test(expected = RedisInvalidSubscriptionException.class) + public void testPSubscribeNotAlive() { + subscription.subscribe(new byte[][] { "a".getBytes() }); + subscription.unsubscribe(); + assertFalse(subscription.isAlive()); + subscription.pSubscribe(new byte[][] { "s*".getBytes() }); + } + + @Test + public void testDoCloseNotSubscribed() { + subscription.doClose(); + verify(pubsub, never()).unsubscribe(new byte[0]); + verify(pubsub, never()).punsubscribe(new byte[0]); + } + + @Test + public void testDoCloseSubscribedChannels() { + subscription.subscribe(new byte[][] { "a".getBytes() }); + subscription.doClose(); + verify(pubsub, times(1)).unsubscribe(new byte[0]); + verify(pubsub, never()).punsubscribe(new byte[0]); + } + + @Test + public void testDoCloseSubscribedPatterns() { + subscription.pSubscribe(new byte[][] { "a*".getBytes() }); + subscription.doClose(); + verify(pubsub, never()).unsubscribe(new byte[0]); + verify(pubsub, times(1)).punsubscribe(new byte[0]); + } + +} diff --git a/src/test/java/org/springframework/data/redis/connection/rjc/RjcSubscriptionTests.java b/src/test/java/org/springframework/data/redis/connection/rjc/RjcSubscriptionTests.java new file mode 100644 index 000000000..c394591aa --- /dev/null +++ b/src/test/java/org/springframework/data/redis/connection/rjc/RjcSubscriptionTests.java @@ -0,0 +1,310 @@ +/* + * Copyright 2011-2013 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.redis.connection.rjc; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +import java.util.Collection; + +import org.idevlab.rjc.message.RedisNodeSubscriber; +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; + +/** + * Unit test of {@link RjcSubscription} + * + * @author Jennifer Hickey + * + */ +public class RjcSubscriptionTests { + + private RjcSubscription subscription; + + private RedisNodeSubscriber subscriber; + + private MessageListener listener; + + @Before + public void setUp() { + subscriber = Mockito.mock(RedisNodeSubscriber.class); + listener = Mockito.mock(MessageListener.class); + subscription = new RjcSubscription(listener, subscriber); + } + + @Test + public void testUnsubscribeAllAndClose() { + byte[][] channel = new byte[][] { "a".getBytes() }; + subscription.subscribe(channel); + subscription.unsubscribe(); + verify(subscriber, never()).close(); + verify(subscriber, times(1)).unsubscribe(RjcUtils.decodeMultiple(channel)); + assertFalse(subscription.isAlive()); + assertTrue(subscription.getChannels().isEmpty()); + assertTrue(subscription.getPatterns().isEmpty()); + } + + @Test + public void testUnsubscribeAllChannelsWithPatterns() { + byte[][] channel = new byte[][] { "a".getBytes() }; + subscription.subscribe(channel); + subscription.pSubscribe(new byte[][] { "s*".getBytes() }); + subscription.unsubscribe(); + verify(subscriber, times(1)).unsubscribe(RjcUtils.decodeMultiple(channel)); + verify(subscriber, never()).close(); + assertTrue(subscription.isAlive()); + assertTrue(subscription.getChannels().isEmpty()); + Collection patterns = subscription.getPatterns(); + assertEquals(1, patterns.size()); + assertArrayEquals("s*".getBytes(), patterns.iterator().next()); + } + + @Test + public void testUnsubscribeChannelAndClose() { + byte[][] channel = new byte[][] { "a".getBytes() }; + subscription.subscribe(channel); + subscription.unsubscribe(channel); + verify(subscriber, times(1)).unsubscribe(RjcUtils.decodeMultiple(channel)); + verify(subscriber, never()).close(); + assertFalse(subscription.isAlive()); + assertTrue(subscription.getChannels().isEmpty()); + assertTrue(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(subscriber, times(1)).unsubscribe( + RjcUtils.decodeMultiple(new byte[][] { "a".getBytes() })); + verify(subscriber, never()).close(); + assertTrue(subscription.isAlive()); + Collection subChannels = subscription.getChannels(); + assertEquals(1, subChannels.size()); + assertArrayEquals("b".getBytes(), subChannels.iterator().next()); + assertTrue(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(subscriber, times(1)).unsubscribe(RjcUtils.decodeMultiple(channel)); + verify(subscriber, never()).close(); + assertTrue(subscription.isAlive()); + assertTrue(subscription.getChannels().isEmpty()); + Collection patterns = subscription.getPatterns(); + assertEquals(1, patterns.size()); + assertArrayEquals("s*".getBytes(), patterns.iterator().next()); + } + + @Test + public void testUnsubscribeChannelWithPatternsSomeLeft() { + byte[][] channel = new byte[][] { "a".getBytes() }; + subscription.subscribe(new byte[][] { "a".getBytes(), "b".getBytes() }); + subscription.pSubscribe(new byte[][] { "s*".getBytes() }); + subscription.unsubscribe(channel); + verify(subscriber, times(1)).unsubscribe(RjcUtils.decodeMultiple(channel)); + verify(subscriber, never()).close(); + assertTrue(subscription.isAlive()); + Collection channels = subscription.getChannels(); + assertEquals(1, channels.size()); + assertArrayEquals("b".getBytes(), channels.iterator().next()); + Collection patterns = subscription.getPatterns(); + assertEquals(1, patterns.size()); + assertArrayEquals("s*".getBytes(), patterns.iterator().next()); + } + + @Test + public void testUnsubscribeAllNoChannels() { + subscription.pSubscribe(new byte[][] { "s*".getBytes() }); + subscription.unsubscribe(); + verify(subscriber, never()).close(); + assertTrue(subscription.isAlive()); + assertTrue(subscription.getChannels().isEmpty()); + Collection patterns = subscription.getPatterns(); + assertEquals(1, patterns.size()); + assertArrayEquals("s*".getBytes(), patterns.iterator().next()); + } + + @Test + public void testUnsubscribeNotAlive() { + byte[][] channel = new byte[][] { "a".getBytes() }; + subscription.subscribe(channel); + subscription.unsubscribe(); + assertFalse(subscription.isAlive()); + subscription.unsubscribe(); + verify(subscriber, times(1)).unsubscribe(RjcUtils.decodeMultiple(channel)); + verify(subscriber, never()).close(); + } + + @Test(expected = RedisInvalidSubscriptionException.class) + public void testSubscribeNotAlive() { + subscription.subscribe(new byte[][] { "a".getBytes() }); + subscription.unsubscribe(); + assertFalse(subscription.isAlive()); + subscription.subscribe(new byte[][] { "s".getBytes() }); + } + + @Test + public void testPUnsubscribeAllAndClose() { + byte[][] pattern = new byte[][] { "a*".getBytes() }; + subscription.pSubscribe(pattern); + subscription.pUnsubscribe(); + verify(subscriber, never()).close(); + verify(subscriber, times(1)).punsubscribe(RjcUtils.decodeMultiple(pattern)); + assertFalse(subscription.isAlive()); + assertTrue(subscription.getChannels().isEmpty()); + assertTrue(subscription.getPatterns().isEmpty()); + } + + @Test + public void testPUnsubscribeAllPatternsWithChannels() { + subscription.subscribe(new byte[][] { "a".getBytes() }); + byte[][] patterns = new byte[][] { "s*".getBytes() }; + subscription.pSubscribe(patterns); + subscription.pUnsubscribe(); + verify(subscriber, never()).close(); + verify(subscriber, times(1)).punsubscribe(RjcUtils.decodeMultiple(patterns)); + assertTrue(subscription.isAlive()); + assertTrue(subscription.getPatterns().isEmpty()); + Collection channels = subscription.getChannels(); + assertEquals(1, channels.size()); + assertArrayEquals("a".getBytes(), channels.iterator().next()); + } + + @Test + public void testPUnsubscribeAndClose() { + byte[][] pattern = new byte[][] { "a*".getBytes() }; + subscription.pSubscribe(pattern); + subscription.pUnsubscribe(pattern); + verify(subscriber, never()).close(); + verify(subscriber, times(1)).punsubscribe(RjcUtils.decodeMultiple(pattern)); + assertFalse(subscription.isAlive()); + assertTrue(subscription.getChannels().isEmpty()); + assertTrue(subscription.getPatterns().isEmpty()); + } + + @Test + public void testPUnsubscribePatternSomeLeft() { + byte[][] patterns = new byte[][] { "a*".getBytes(), "b*".getBytes() }; + subscription.pSubscribe(patterns); + byte[][] pattern = new byte[][] { "a*".getBytes() }; + subscription.pUnsubscribe(pattern); + verify(subscriber, times(1)).punsubscribe(RjcUtils.decodeMultiple(pattern)); + verify(subscriber, never()).close(); + assertTrue(subscription.isAlive()); + Collection subPatterns = subscription.getPatterns(); + assertEquals(1, subPatterns.size()); + assertArrayEquals("b*".getBytes(), subPatterns.iterator().next()); + assertTrue(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(subscriber, times(1)).punsubscribe(RjcUtils.decodeMultiple(pattern)); + verify(subscriber, never()).close(); + assertTrue(subscription.isAlive()); + assertTrue(subscription.getPatterns().isEmpty()); + Collection channels = subscription.getChannels(); + assertEquals(1, channels.size()); + assertArrayEquals("a".getBytes(), channels.iterator().next()); + } + + @Test + public void testUnsubscribePatternWithChannelsSomeLeft() { + byte[][] pattern = new byte[][] { "a*".getBytes() }; + subscription.pSubscribe(new byte[][] { "a*".getBytes(), "b*".getBytes() }); + subscription.subscribe(new byte[][] { "a".getBytes() }); + subscription.pUnsubscribe(pattern); + verify(subscriber, never()).close(); + verify(subscriber, times(1)).punsubscribe(RjcUtils.decodeMultiple(pattern)); + assertTrue(subscription.isAlive()); + Collection channels = subscription.getChannels(); + assertEquals(1, channels.size()); + assertArrayEquals("a".getBytes(), channels.iterator().next()); + Collection patterns = subscription.getPatterns(); + assertEquals(1, patterns.size()); + assertArrayEquals("b*".getBytes(), patterns.iterator().next()); + } + + @Test + public void testPUnsubscribeAllNoPatterns() { + subscription.subscribe(new byte[][] { "s".getBytes() }); + subscription.pUnsubscribe(); + verify(subscriber, never()).close(); + assertTrue(subscription.isAlive()); + assertTrue(subscription.getPatterns().isEmpty()); + Collection channels = subscription.getChannels(); + assertEquals(1, channels.size()); + assertArrayEquals("s".getBytes(), channels.iterator().next()); + } + + @Test + public void testPUnsubscribeNotAlive() { + byte[][] channels = new byte[][] { "a".getBytes() }; + subscription.subscribe(channels); + subscription.unsubscribe(); + assertFalse(subscription.isAlive()); + subscription.pUnsubscribe(); + verify(subscriber, times(1)).unsubscribe(RjcUtils.decodeMultiple(channels)); + verify(subscriber, never()).close(); + } + + @Test(expected = RedisInvalidSubscriptionException.class) + public void testPSubscribeNotAlive() { + subscription.subscribe(new byte[][] { "a".getBytes() }); + subscription.unsubscribe(); + assertFalse(subscription.isAlive()); + subscription.pSubscribe(new byte[][] { "s*".getBytes() }); + } + + @Test + public void testDoCloseNotSubscribed() { + subscription.doClose(); + verify(subscriber, never()).close(); + } + + @Test + public void testDoCloseSubscribedChannels() { + subscription.subscribe(new byte[][] { "a".getBytes() }); + subscription.doClose(); + verify(subscriber, times(1)).close(); + } + + @Test + public void testDoCloseSubscribedPatterns() { + subscription.pSubscribe(new byte[][] { "a*".getBytes() }); + subscription.doClose(); + verify(subscriber, times(1)).close(); + } + +} diff --git a/src/test/java/org/springframework/data/redis/connection/srp/SrpSubscriptionTests.java b/src/test/java/org/springframework/data/redis/connection/srp/SrpSubscriptionTests.java new file mode 100644 index 000000000..d49583994 --- /dev/null +++ b/src/test/java/org/springframework/data/redis/connection/srp/SrpSubscriptionTests.java @@ -0,0 +1,315 @@ +/* + * Copyright 2011-2013 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.redis.connection.srp; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.any; +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; + +import redis.client.RedisClient; +import redis.client.ReplyListener; + +/** + * Unit test of {@link SrpSubscription} + * + * @author Jennifer Hickey + * + */ +public class SrpSubscriptionTests { + + private SrpSubscription subscription; + + private RedisClient redisClient; + + private MessageListener listener; + + @Before + public void setUp() { + redisClient = Mockito.mock(RedisClient.class); + listener = Mockito.mock(MessageListener.class); + subscription = new SrpSubscription(listener, redisClient); + } + + @Test + public void testUnsubscribeAllAndClose() { + subscription.subscribe(new byte[][] { "a".getBytes() }); + subscription.unsubscribe(); + verify(redisClient, times(1)).unsubscribe((Object[]) null); + verify(redisClient, never()).punsubscribe((Object[]) null); + assertFalse(subscription.isAlive()); + verify(redisClient).removeListener(any(ReplyListener.class)); + assertTrue(subscription.getChannels().isEmpty()); + assertTrue(subscription.getPatterns().isEmpty()); + } + + @Test + public void testUnsubscribeAllChannelsWithPatterns() { + subscription.subscribe(new byte[][] { "a".getBytes() }); + subscription.pSubscribe(new byte[][] { "s*".getBytes() }); + subscription.unsubscribe(); + verify(redisClient, times(1)).unsubscribe((Object[]) null); + verify(redisClient, never()).punsubscribe((Object[]) null); + assertTrue(subscription.isAlive()); + assertTrue(subscription.getChannels().isEmpty()); + Collection patterns = subscription.getPatterns(); + assertEquals(1, patterns.size()); + assertArrayEquals("s*".getBytes(), patterns.iterator().next()); + } + + @Test + public void testUnsubscribeChannelAndClose() { + byte[][] channel = new byte[][] { "a".getBytes() }; + subscription.subscribe(channel); + subscription.unsubscribe(channel); + verify(redisClient, times(1)).unsubscribe((Object[]) channel); + verify(redisClient, never()).punsubscribe((Object[]) null); + verify(redisClient).removeListener(any(ReplyListener.class)); + assertFalse(subscription.isAlive()); + assertTrue(subscription.getChannels().isEmpty()); + assertTrue(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(redisClient, times(1)).unsubscribe((Object[]) new byte[][] { "a".getBytes() }); + verify(redisClient, never()).punsubscribe((Object[]) null); + assertTrue(subscription.isAlive()); + Collection subChannels = subscription.getChannels(); + assertEquals(1, subChannels.size()); + assertArrayEquals("b".getBytes(), subChannels.iterator().next()); + assertTrue(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(redisClient, times(1)).unsubscribe((Object[]) channel); + verify(redisClient, never()).punsubscribe((Object[]) null); + assertTrue(subscription.isAlive()); + assertTrue(subscription.getChannels().isEmpty()); + Collection patterns = subscription.getPatterns(); + assertEquals(1, patterns.size()); + assertArrayEquals("s*".getBytes(), patterns.iterator().next()); + } + + @Test + public void testUnsubscribeChannelWithPatternsSomeLeft() { + byte[][] channel = new byte[][] { "a".getBytes() }; + subscription.subscribe(new byte[][] { "a".getBytes(), "b".getBytes() }); + subscription.pSubscribe(new byte[][] { "s*".getBytes() }); + subscription.unsubscribe(channel); + verify(redisClient, times(1)).unsubscribe((Object[]) channel); + verify(redisClient, never()).punsubscribe((Object[]) null); + assertTrue(subscription.isAlive()); + Collection channels = subscription.getChannels(); + assertEquals(1, channels.size()); + assertArrayEquals("b".getBytes(), channels.iterator().next()); + Collection patterns = subscription.getPatterns(); + assertEquals(1, patterns.size()); + assertArrayEquals("s*".getBytes(), patterns.iterator().next()); + } + + @Test + public void testUnsubscribeAllNoChannels() { + subscription.pSubscribe(new byte[][] { "s*".getBytes() }); + subscription.unsubscribe(); + verify(redisClient, never()).unsubscribe((Object[]) null); + verify(redisClient, never()).punsubscribe((Object[]) null); + assertTrue(subscription.isAlive()); + assertTrue(subscription.getChannels().isEmpty()); + Collection patterns = subscription.getPatterns(); + assertEquals(1, patterns.size()); + assertArrayEquals("s*".getBytes(), patterns.iterator().next()); + } + + @Test + public void testUnsubscribeNotAlive() { + subscription.subscribe(new byte[][] { "a".getBytes() }); + subscription.unsubscribe(); + assertFalse(subscription.isAlive()); + subscription.unsubscribe(); + verify(redisClient, times(1)).removeListener(any(ReplyListener.class)); + verify(redisClient, times(1)).unsubscribe((Object[]) null); + verify(redisClient, never()).punsubscribe((Object[]) null); + } + + @Test(expected = RedisInvalidSubscriptionException.class) + public void testSubscribeNotAlive() { + subscription.subscribe(new byte[][] { "a".getBytes() }); + subscription.unsubscribe(); + assertFalse(subscription.isAlive()); + subscription.subscribe(new byte[][] { "s".getBytes() }); + } + + @Test + public void testPUnsubscribeAllAndClose() { + subscription.pSubscribe(new byte[][] { "a*".getBytes() }); + subscription.pUnsubscribe(); + verify(redisClient, never()).unsubscribe((Object[]) null); + verify(redisClient, times(1)).punsubscribe((Object[]) null); + assertFalse(subscription.isAlive()); + verify(redisClient).removeListener(any(ReplyListener.class)); + assertTrue(subscription.getChannels().isEmpty()); + assertTrue(subscription.getPatterns().isEmpty()); + } + + @Test + public void testPUnsubscribeAllPatternsWithChannels() { + subscription.subscribe(new byte[][] { "a".getBytes() }); + subscription.pSubscribe(new byte[][] { "s*".getBytes() }); + subscription.pUnsubscribe(); + verify(redisClient, never()).unsubscribe((Object[]) null); + verify(redisClient, times(1)).punsubscribe((Object[]) null); + assertTrue(subscription.isAlive()); + assertTrue(subscription.getPatterns().isEmpty()); + Collection channels = subscription.getChannels(); + assertEquals(1, channels.size()); + assertArrayEquals("a".getBytes(), channels.iterator().next()); + } + + @Test + public void testPUnsubscribeAndClose() { + byte[][] pattern = new byte[][] { "a*".getBytes() }; + subscription.pSubscribe(pattern); + subscription.pUnsubscribe(pattern); + verify(redisClient, never()).unsubscribe((Object[]) null); + verify(redisClient, times(1)).punsubscribe((Object[]) pattern); + assertFalse(subscription.isAlive()); + verify(redisClient).removeListener(any(ReplyListener.class)); + assertTrue(subscription.getChannels().isEmpty()); + assertTrue(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(redisClient, times(1)).punsubscribe((Object[]) new byte[][] { "a*".getBytes() }); + verify(redisClient, never()).unsubscribe((Object[]) null); + assertTrue(subscription.isAlive()); + Collection subPatterns = subscription.getPatterns(); + assertEquals(1, subPatterns.size()); + assertArrayEquals("b*".getBytes(), subPatterns.iterator().next()); + assertTrue(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(redisClient, times(1)).punsubscribe((Object[]) pattern); + verify(redisClient, never()).unsubscribe((Object[]) null); + assertTrue(subscription.isAlive()); + assertTrue(subscription.getPatterns().isEmpty()); + Collection channels = subscription.getChannels(); + assertEquals(1, channels.size()); + assertArrayEquals("a".getBytes(), channels.iterator().next()); + } + + @Test + public void testUnsubscribePatternWithChannelsSomeLeft() { + byte[][] pattern = new byte[][] { "a*".getBytes() }; + subscription.pSubscribe(new byte[][] { "a*".getBytes(), "b*".getBytes() }); + subscription.subscribe(new byte[][] { "a".getBytes() }); + subscription.pUnsubscribe(pattern); + verify(redisClient, never()).unsubscribe((Object[]) null); + verify(redisClient, times(1)).punsubscribe((Object[]) pattern); + assertTrue(subscription.isAlive()); + Collection channels = subscription.getChannels(); + assertEquals(1, channels.size()); + assertArrayEquals("a".getBytes(), channels.iterator().next()); + Collection patterns = subscription.getPatterns(); + assertEquals(1, patterns.size()); + assertArrayEquals("b*".getBytes(), patterns.iterator().next()); + } + + @Test + public void testPUnsubscribeAllNoPatterns() { + subscription.subscribe(new byte[][] { "s".getBytes() }); + subscription.pUnsubscribe(); + verify(redisClient, never()).unsubscribe((Object[]) null); + verify(redisClient, never()).punsubscribe((Object[]) null); + assertTrue(subscription.isAlive()); + assertTrue(subscription.getPatterns().isEmpty()); + Collection channels = subscription.getChannels(); + assertEquals(1, channels.size()); + assertArrayEquals("s".getBytes(), channels.iterator().next()); + } + + @Test + public void testPUnsubscribeNotAlive() { + subscription.subscribe(new byte[][] { "a".getBytes() }); + subscription.unsubscribe(); + assertFalse(subscription.isAlive()); + subscription.pUnsubscribe(); + verify(redisClient, times(1)).unsubscribe((Object[]) null); + verify(redisClient, never()).punsubscribe((Object[]) null); + verify(redisClient, times(1)).removeListener(any(ReplyListener.class)); + } + + @Test(expected = RedisInvalidSubscriptionException.class) + public void testPSubscribeNotAlive() { + subscription.subscribe(new byte[][] { "a".getBytes() }); + subscription.unsubscribe(); + assertFalse(subscription.isAlive()); + subscription.pSubscribe(new byte[][] { "s*".getBytes() }); + } + + @Test + public void testDoCloseNotSubscribed() { + subscription.doClose(); + verify(redisClient, never()).unsubscribe((Object[]) null); + verify(redisClient, never()).punsubscribe((Object[]) null); + } + + @Test + public void testDoCloseSubscribedChannels() { + subscription.subscribe(new byte[][] { "a".getBytes() }); + subscription.doClose(); + verify(redisClient, times(1)).unsubscribe((Object[]) null); + verify(redisClient, never()).punsubscribe((Object[]) null); + } + + @Test + public void testDoCloseSubscribedPatterns() { + subscription.pSubscribe(new byte[][] { "a*".getBytes() }); + subscription.doClose(); + verify(redisClient, never()).unsubscribe((Object[]) null); + verify(redisClient, times(1)).punsubscribe((Object[]) null); + } + +} diff --git a/src/test/java/org/springframework/data/redis/listener/SubscriptionConnectionTests.java b/src/test/java/org/springframework/data/redis/listener/SubscriptionConnectionTests.java index c85550afb..87f703dde 100644 --- a/src/test/java/org/springframework/data/redis/listener/SubscriptionConnectionTests.java +++ b/src/test/java/org/springframework/data/redis/listener/SubscriptionConnectionTests.java @@ -17,21 +17,28 @@ package org.springframework.data.redis.listener; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collection; import java.util.List; import org.junit.After; -import org.junit.Before; +import org.junit.AfterClass; import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; +import org.junit.runners.Parameterized.Parameters; import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.core.task.SyncTaskExecutor; +import org.springframework.data.redis.ConnectionFactoryTracker; import org.springframework.data.redis.SettingsUtils; import org.springframework.data.redis.connection.MessageListener; import org.springframework.data.redis.connection.RedisConnection; +import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; +import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory; +import org.springframework.data.redis.connection.rjc.RjcConnectionFactory; +import org.springframework.data.redis.connection.srp.SrpConnectionFactory; import org.springframework.data.redis.listener.adapter.MessageListenerAdapter; -import redis.clients.jedis.JedisPoolConfig; - /** * Integration tests confirming that {@link RedisMessageListenerContainer} * closes connections after unsubscribing @@ -39,11 +46,12 @@ import redis.clients.jedis.JedisPoolConfig; * @author Jennifer Hickey * */ +@RunWith(Parameterized.class) public class SubscriptionConnectionTests { private static final String CHANNEL = "pubsub::test"; - private JedisConnectionFactory connectionFactory; + private RedisConnectionFactory connectionFactory; private List containers = new ArrayList(); @@ -54,17 +62,9 @@ public class SubscriptionConnectionTests { } }; - @Before - public void setUp() { - connectionFactory = new JedisConnectionFactory(); - connectionFactory.setUsePool(true); - connectionFactory.setPort(SettingsUtils.getPort()); - connectionFactory.setHostName(SettingsUtils.getHost()); - connectionFactory.setDatabase(2); - JedisPoolConfig config = new JedisPoolConfig(); - config.setMaxWait(3000l); - connectionFactory.setPoolConfig(config); - connectionFactory.afterPropertiesSet(); + public SubscriptionConnectionTests(RedisConnectionFactory connectionFactory) { + this.connectionFactory = connectionFactory; + ConnectionFactoryTracker.add(connectionFactory); } @After @@ -74,7 +74,45 @@ public class SubscriptionConnectionTests { container.destroy(); } } - connectionFactory.destroy(); + } + + @AfterClass + public static void cleanUp() { + ConnectionFactoryTracker.cleanUp(); + } + + @Parameters + public static Collection testParams() { + // Jedis + JedisConnectionFactory jedisConnFactory = new JedisConnectionFactory(); + jedisConnFactory.setPort(SettingsUtils.getPort()); + jedisConnFactory.setHostName(SettingsUtils.getHost()); + jedisConnFactory.setDatabase(2); + jedisConnFactory.afterPropertiesSet(); + + // Lettuce + LettuceConnectionFactory lettuceConnFactory = new LettuceConnectionFactory(); + lettuceConnFactory.setPort(SettingsUtils.getPort()); + lettuceConnFactory.setHostName(SettingsUtils.getHost()); + lettuceConnFactory.setDatabase(2); + lettuceConnFactory.setValidateConnection(true); + lettuceConnFactory.afterPropertiesSet(); + + // SRP + SrpConnectionFactory srpConnFactory = new SrpConnectionFactory(); + srpConnFactory.setPort(SettingsUtils.getPort()); + srpConnFactory.setHostName(SettingsUtils.getHost()); + srpConnFactory.afterPropertiesSet(); + + // RJC + RjcConnectionFactory rjcConnFactory = new RjcConnectionFactory(); + rjcConnFactory.setPort(SettingsUtils.getPort()); + rjcConnFactory.setHostName(SettingsUtils.getHost()); + rjcConnFactory.setDatabase(2); + rjcConnFactory.afterPropertiesSet(); + + return Arrays.asList(new Object[][] { { jedisConnFactory }, { lettuceConnFactory }, + { srpConnFactory }, { rjcConnFactory } }); } @Test @@ -110,8 +148,7 @@ public class SubscriptionConnectionTests { RedisMessageListenerContainer container = new RedisMessageListenerContainer(); container.setConnectionFactory(connectionFactory); container.setBeanName("container" + i); - container.addMessageListener(listener, - Arrays.asList(new ChannelTopic(CHANNEL))); + container.addMessageListener(listener, Arrays.asList(new ChannelTopic(CHANNEL))); container.setTaskExecutor(new SyncTaskExecutor()); container.setSubscriptionExecutor(new SimpleAsyncTaskExecutor()); container.afterPropertiesSet(); @@ -140,8 +177,7 @@ public class SubscriptionConnectionTests { RedisMessageListenerContainer container = new RedisMessageListenerContainer(); container.setConnectionFactory(connectionFactory); container.setBeanName("container" + i); - container.addMessageListener(listener, - Arrays.asList(new ChannelTopic(CHANNEL))); + container.addMessageListener(listener, Arrays.asList(new ChannelTopic(CHANNEL))); container.setTaskExecutor(new SyncTaskExecutor()); container.setSubscriptionExecutor(new SimpleAsyncTaskExecutor()); container.afterPropertiesSet(); @@ -154,8 +190,7 @@ public class SubscriptionConnectionTests { Thread.sleep(100); // Unsubscribe all listeners from all topics, freeing up a connection - containers.get(0).removeMessageListener(null, - Arrays.asList(new Topic[] {})); + containers.get(0).removeMessageListener(null, Arrays.asList(new Topic[] {})); // verify we can now get a connection from the pool RedisConnection connection = connectionFactory.getConnection();