From 0c9e8f5054d12005dc521cf716c9ff85405469b0 Mon Sep 17 00:00:00 2001 From: Jennifer Hickey Date: Wed, 17 Apr 2013 10:03:09 -0700 Subject: [PATCH] Fix corrupt subscription connections return to pool with Redis 2.6 Redis 2.6 responds to unsubscribe requests even if not subscribed. This results in unread data on the InputStream on Subscription close if extra unsubscribe requests are issued. This causes problems if connection is reused (as in the case of Jedis pool). - Only unsubscribe on Subscription close if still subscribed - Use dedicated connection for RJC subscriptions, as RJC makes extra unsubscribe calls on subscriber close --- .../connection/jedis/JedisConnection.java | 6 - .../connection/jedis/JedisSubscription.java | 8 +- .../lettuce/LettuceSubscription.java | 8 +- .../redis/connection/rjc/RjcConnection.java | 37 +- .../connection/rjc/RjcConnectionFactory.java | 6 +- .../redis/connection/rjc/RjcSubscription.java | 6 +- .../redis/connection/srp/SrpSubscription.java | 8 +- .../connection/util/AbstractSubscription.java | 19 +- .../jedis/JedisSubscriptionTests.java | 307 ++++++++++++++++ .../lettuce/LettuceSubscriptionTests.java | 335 ++++++++++++++++++ .../connection/rjc/RjcSubscriptionTests.java | 310 ++++++++++++++++ .../connection/srp/SrpSubscriptionTests.java | 315 ++++++++++++++++ .../listener/SubscriptionConnectionTests.java | 79 +++-- 13 files changed, 1389 insertions(+), 55 deletions(-) create mode 100644 src/test/java/org/springframework/data/redis/connection/jedis/JedisSubscriptionTests.java create mode 100644 src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSubscriptionTests.java create mode 100644 src/test/java/org/springframework/data/redis/connection/rjc/RjcSubscriptionTests.java create mode 100644 src/test/java/org/springframework/data/redis/connection/srp/SrpSubscriptionTests.java 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();