Spring/spring-data-redis
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DATAREDIS-1173 - Polishing.

Browse Source 
Add Override annotations and Javadoc. Reformat test.

Original Pull Request: #549
This commit is contained in:
Mark Paluch
2020-07-15 14:06:21 +02:00
committed by Christoph Strobl
parent 27c3d0962d
commit 164bdc97fb
4 changed files with 195 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
src/main/java/org/springframework/data/redis/connection/jedis/JedisSubscription.java
View File

@@ -36,6 +36,11 @@ class JedisSubscription extends AbstractSubscription {
this.jedisPubSub = jedisPubSub;
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.util.AbstractSubscription#doClose()
*/
@Override
protected void doClose() {
if (!getChannels().isEmpty()) {
jedisPubSub.unsubscribe();
@@ -45,10 +50,20 @@ class JedisSubscription extends AbstractSubscription {
}
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.util.AbstractSubscription#doPsubscribe(byte[][])
*/
@Override
protected void doPsubscribe(byte[]... patterns) {
jedisPubSub.psubscribe(patterns);
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.util.AbstractSubscription#doPUnsubscribe(boolean, byte[][])
*/
@Override
protected void doPUnsubscribe(boolean all, byte[]... patterns) {
if (all) {
jedisPubSub.punsubscribe();
@@ -57,10 +72,20 @@ class JedisSubscription extends AbstractSubscription {
}
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.util.AbstractSubscription#doSubscribe(byte[][])
*/
@Override
protected void doSubscribe(byte[]... channels) {
jedisPubSub.subscribe(channels);
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.util.AbstractSubscription#doUnsubscribe(boolean, byte[][])
*/
@Override
protected void doUnsubscribe(boolean all, byte[]... channels) {
if (all) {
jedisPubSub.unsubscribe();

6
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceSubscription.java
View File

@@ -67,6 +67,7 @@ public class LettuceSubscription extends AbstractSubscription {
* (non-Javadoc)
* @see org.springframework.data.redis.connection.util.AbstractSubscription#doClose()
*/
@Override
protected void doClose() {
if (!getChannels().isEmpty()) {
@@ -85,6 +86,7 @@ public class LettuceSubscription extends AbstractSubscription {
* (non-Javadoc)
* @see org.springframework.data.redis.connection.util.AbstractSubscription#doPsubscribe(byte[][])
*/
@Override
protected void doPsubscribe(byte[]... patterns) {
pubsub.psubscribe(patterns);
}
@@ -93,6 +95,7 @@ public class LettuceSubscription extends AbstractSubscription {
* (non-Javadoc)
* @see org.springframework.data.redis.connection.util.AbstractSubscription#doPUnsubscribe(boolean, byte[][])
*/
@Override
protected void doPUnsubscribe(boolean all, byte[]... patterns) {
if (all) {
@@ -106,6 +109,7 @@ public class LettuceSubscription extends AbstractSubscription {
* (non-Javadoc)
* @see org.springframework.data.redis.connection.util.AbstractSubscription#doSubscribe(byte[][])
*/
@Override
protected void doSubscribe(byte[]... channels) {
pubsub.subscribe(channels);
}
@@ -114,6 +118,7 @@ public class LettuceSubscription extends AbstractSubscription {
* (non-Javadoc)
* @see org.springframework.data.redis.connection.util.AbstractSubscription#doUnsubscribe(boolean, byte[][])
*/
@Override
protected void doUnsubscribe(boolean all, byte[]... channels) {
if (all) {
@@ -122,5 +127,4 @@ public class LettuceSubscription extends AbstractSubscription {
pubsub.unsubscribe(channels);
}
}
}

50
src/main/java/org/springframework/data/redis/connection/util/AbstractSubscription.java
View File

@@ -111,22 +111,42 @@ public abstract class AbstractSubscription implements Subscription {
*/
protected abstract void doClose();
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.Subscription#getListener()
*/
@Override
public MessageListener getListener() {
return listener;
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.Subscription#getChannels()
*/
@Override
public Collection<byte[]> getChannels() {
synchronized (channels) {
return clone(channels);
}
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.Subscription#getPatterns()
*/
@Override
public Collection<byte[]> getPatterns() {
synchronized (patterns) {
return clone(patterns);
}
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.Subscription#pSubscribe(byte[][])
*/
@Override
public void pSubscribe(byte[]... patterns) {
checkPulse();
@@ -139,10 +159,20 @@ public abstract class AbstractSubscription implements Subscription {
doPsubscribe(patterns);
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.Subscription#pUnsubscribe()
*/
@Override
public void pUnsubscribe() {
pUnsubscribe((byte[][]) null);
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.Subscription#subscribe(byte[][])
*/
@Override
public void subscribe(byte[]... channels) {
checkPulse();
@@ -155,10 +185,20 @@ public abstract class AbstractSubscription implements Subscription {
doSubscribe(channels);
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.Subscription#unsubscribe()
*/
@Override
public void unsubscribe() {
unsubscribe((byte[][]) null);
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.Subscription#pUnsubscribe(byte[][])
*/
@Override
public void pUnsubscribe(@Nullable byte[]... patts) {
if (!isAlive()) {
return;
@@ -186,6 +226,11 @@ public abstract class AbstractSubscription implements Subscription {
closeIfUnsubscribed();
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.Subscription#unsubscribe(byte[][])
*/
@Override
public void unsubscribe(@Nullable byte[]... chans) {
if (!isAlive()) {
return;
@@ -213,6 +258,11 @@ public abstract class AbstractSubscription implements Subscription {
closeIfUnsubscribed();
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.Subscription#isAlive()
*/
@Override
public boolean isAlive() {
return alive.get();
}

156
src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSubscriptionTests.java
View File

@@ -25,7 +25,6 @@ import java.util.Collection;
import org.junit.Before;
import org.junit.Test;
import org.mockito.Mockito;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.connection.RedisInvalidSubscriptionException;
@@ -41,9 +40,7 @@ public class LettuceSubscriptionTests {
private LettuceSubscription subscription;
StatefulRedisPubSubConnection<byte[], byte[]> pubsub;
private MessageListener listener;
private StatefulRedisPubSubConnection<byte[], byte[]> pubsub;
private RedisPubSubCommands<byte[], byte[]> asyncCommands;
@@ -53,126 +50,159 @@ public class LettuceSubscriptionTests {
@Before
public void setUp() {
pubsub = Mockito.mock(StatefulRedisPubSubConnection.class);
listener = Mockito.mock(MessageListener.class);
asyncCommands = Mockito.mock(RedisPubSubCommands.class);
connectionProvider = Mockito.mock(LettuceConnectionProvider.class);
pubsub = mock(StatefulRedisPubSubConnection.class);
asyncCommands = mock(RedisPubSubCommands.class);
connectionProvider = mock(LettuceConnectionProvider.class);
Mockito.when(pubsub.sync()).thenReturn(asyncCommands);
subscription = new LettuceSubscription(listener, pubsub, connectionProvider);
when(pubsub.sync()).thenReturn(asyncCommands);
subscription = new LettuceSubscription(mock(MessageListener.class), pubsub, connectionProvider);
}
@Test
public void testUnsubscribeAllAndClose() {
subscription.subscribe(new byte[][] { "a".getBytes() });
subscription.unsubscribe();
verify(asyncCommands).unsubscribe();
verify(asyncCommands, never()).punsubscribe();
verify(connectionProvider).release(pubsub);
verify(pubsub).removeListener(any(LettuceMessageListener.class));
assertThat(subscription.isAlive()).isFalse();
assertThat(subscription.getChannels().isEmpty()).isTrue();
assertThat(subscription.getPatterns().isEmpty()).isTrue();
assertThat(subscription.getChannels()).isEmpty();
assertThat(subscription.getPatterns()).isEmpty();
}
@Test
public void testUnsubscribeAllChannelsWithPatterns() {
subscription.subscribe(new byte[][] { "a".getBytes() });
subscription.pSubscribe(new byte[][] { "s*".getBytes() });
subscription.unsubscribe();
verify(asyncCommands).unsubscribe();
verify(asyncCommands, never()).punsubscribe();
assertThat(subscription.isAlive()).isTrue();
assertThat(subscription.getChannels().isEmpty()).isTrue();
assertThat(subscription.getChannels()).isEmpty();
Collection<byte[]> patterns = subscription.getPatterns();
assertThat(patterns.size()).isEqualTo(1);
assertThat(patterns).hasSize(1);
assertThat(patterns.iterator().next()).isEqualTo("s*".getBytes());
}
@Test
public void testUnsubscribeChannelAndClose() {
byte[][] channel = new byte[][] { "a".getBytes() };
subscription.subscribe(channel);
subscription.unsubscribe(channel);
verify(asyncCommands).unsubscribe(channel);
verify(asyncCommands, never()).unsubscribe();
verify(asyncCommands, never()).punsubscribe();
verify(connectionProvider).release(pubsub);
verify(pubsub).removeListener(any(LettuceMessageListener.class));
assertThat(subscription.isAlive()).isFalse();
assertThat(subscription.getChannels().isEmpty()).isTrue();
assertThat(subscription.getPatterns().isEmpty()).isTrue();
assertThat(subscription.getChannels()).isEmpty();
assertThat(subscription.getPatterns()).isEmpty();
}
@Test
public void testUnsubscribeChannelSomeLeft() {
byte[][] channels = new byte[][] { "a".getBytes(), "b".getBytes() };
subscription.subscribe(channels);
subscription.unsubscribe(new byte[][] { "a".getBytes() });
verify(asyncCommands).unsubscribe(new byte[][] { "a".getBytes() });
verify(asyncCommands, never()).unsubscribe();
verify(asyncCommands, never()).punsubscribe();
assertThat(subscription.isAlive()).isTrue();
Collection<byte[]> subChannels = subscription.getChannels();
assertThat(subChannels.size()).isEqualTo(1);
assertThat(subChannels).hasSize(1);
assertThat(subChannels.iterator().next()).isEqualTo("b".getBytes());
assertThat(subscription.getPatterns().isEmpty()).isTrue();
assertThat(subscription.getPatterns()).isEmpty();
}
@Test
public void testUnsubscribeChannelWithPatterns() {
byte[][] channel = new byte[][] { "a".getBytes() };
subscription.subscribe(channel);
subscription.pSubscribe(new byte[][] { "s*".getBytes() });
subscription.unsubscribe(channel);
verify(asyncCommands).unsubscribe(channel);
verify(asyncCommands, never()).unsubscribe();
verify(asyncCommands, never()).punsubscribe();
assertThat(subscription.isAlive()).isTrue();
assertThat(subscription.getChannels().isEmpty()).isTrue();
assertThat(subscription.getChannels()).isEmpty();
Collection<byte[]> patterns = subscription.getPatterns();
assertThat(patterns.size()).isEqualTo(1);
assertThat(patterns).hasSize(1);
assertThat(patterns.iterator().next()).isEqualTo("s*".getBytes());
}
@Test
public void testUnsubscribeChannelWithPatternsSomeLeft() {
byte[][] channel = new byte[][] { "a".getBytes() };
subscription.subscribe(new byte[][] { "a".getBytes(), "b".getBytes() });
subscription.subscribe("a".getBytes(), "b".getBytes());
subscription.pSubscribe(new byte[][] { "s*".getBytes() });
subscription.unsubscribe(channel);
verify(asyncCommands).unsubscribe(channel);
verify(asyncCommands, never()).unsubscribe();
verify(asyncCommands, never()).punsubscribe();
assertThat(subscription.isAlive()).isTrue();
Collection<byte[]> channels = subscription.getChannels();
assertThat(channels.size()).isEqualTo(1);
assertThat(channels).hasSize(1);
assertThat(channels.iterator().next()).isEqualTo("b".getBytes());
Collection<byte[]> patterns = subscription.getPatterns();
assertThat(patterns.size()).isEqualTo(1);
assertThat(patterns).hasSize(1);
assertThat(patterns.iterator().next()).isEqualTo("s*".getBytes());
}
@Test
public void testUnsubscribeAllNoChannels() {
subscription.pSubscribe(new byte[][] { "s*".getBytes() });
subscription.unsubscribe();
verify(asyncCommands, never()).unsubscribe();
verify(asyncCommands, never()).punsubscribe();
assertThat(subscription.isAlive()).isTrue();
assertThat(subscription.getChannels().isEmpty()).isTrue();
assertThat(subscription.getChannels()).isEmpty();
Collection<byte[]> patterns = subscription.getPatterns();
assertThat(patterns.size()).isEqualTo(1);
assertThat(patterns).hasSize(1);
assertThat(patterns.iterator().next()).isEqualTo("s*".getBytes());
}
@Test
public void testUnsubscribeNotAlive() {
subscription.subscribe(new byte[][] { "a".getBytes() });
subscription.unsubscribe();
verify(connectionProvider).release(pubsub);
verify(pubsub).removeListener(any(LettuceMessageListener.class));
assertThat(subscription.isAlive()).isFalse();
subscription.unsubscribe();
verify(asyncCommands).unsubscribe();
verify(asyncCommands, never()).punsubscribe();
@@ -180,122 +210,157 @@ public class LettuceSubscriptionTests {
@Test(expected = RedisInvalidSubscriptionException.class)
public void testSubscribeNotAlive() {
subscription.subscribe(new byte[][] { "a".getBytes() });
subscription.unsubscribe();
assertThat(subscription.isAlive()).isFalse();
subscription.subscribe(new byte[][] { "s".getBytes() });
}
@Test
public void testPUnsubscribeAllAndClose() {
subscription.pSubscribe(new byte[][] { "a*".getBytes() });
subscription.pUnsubscribe();
verify(asyncCommands, never()).unsubscribe();
verify(asyncCommands).punsubscribe();
assertThat(subscription.isAlive()).isFalse();
verify(connectionProvider).release(pubsub);
verify(pubsub).removeListener(any(LettuceMessageListener.class));
assertThat(subscription.getChannels().isEmpty()).isTrue();
assertThat(subscription.getPatterns().isEmpty()).isTrue();
assertThat(subscription.isAlive()).isFalse();
assertThat(subscription.getChannels()).isEmpty();
assertThat(subscription.getPatterns()).isEmpty();
}
@Test
public void testPUnsubscribeAllPatternsWithChannels() {
subscription.subscribe(new byte[][] { "a".getBytes() });
subscription.pSubscribe(new byte[][] { "s*".getBytes() });
subscription.pUnsubscribe();
verify(asyncCommands, never()).unsubscribe();
verify(asyncCommands).punsubscribe();
assertThat(subscription.isAlive()).isTrue();
assertThat(subscription.getPatterns().isEmpty()).isTrue();
assertThat(subscription.getPatterns()).isEmpty();
Collection<byte[]> channels = subscription.getChannels();
assertThat(channels.size()).isEqualTo(1);
assertThat(channels).hasSize(1);
assertThat(channels.iterator().next()).isEqualTo("a".getBytes());
}
@Test
public void testPUnsubscribeAndClose() {
byte[][] pattern = new byte[][] { "a*".getBytes() };
subscription.pSubscribe(pattern);
subscription.pUnsubscribe(pattern);
verify(asyncCommands, never()).unsubscribe();
verify(asyncCommands, never()).punsubscribe();
verify(asyncCommands).punsubscribe(pattern);
verify(connectionProvider).release(pubsub);
verify(pubsub).removeListener(any(LettuceMessageListener.class));
assertThat(subscription.isAlive()).isFalse();
assertThat(subscription.getChannels().isEmpty()).isTrue();
assertThat(subscription.getPatterns().isEmpty()).isTrue();
assertThat(subscription.getChannels()).isEmpty();
assertThat(subscription.getPatterns()).isEmpty();
}
@Test
public void testPUnsubscribePatternSomeLeft() {
byte[][] patterns = new byte[][] { "a*".getBytes(), "b*".getBytes() };
subscription.pSubscribe(patterns);
subscription.pUnsubscribe(new byte[][] { "a*".getBytes() });
verify(asyncCommands).punsubscribe(new byte[][] { "a*".getBytes() });
verify(asyncCommands, never()).unsubscribe();
verify(asyncCommands, never()).punsubscribe();
assertThat(subscription.isAlive()).isTrue();
Collection<byte[]> subPatterns = subscription.getPatterns();
assertThat(subPatterns.size()).isEqualTo(1);
assertThat(subPatterns).hasSize(1);
assertThat(subPatterns.iterator().next()).isEqualTo("b*".getBytes());
assertThat(subscription.getChannels().isEmpty()).isTrue();
assertThat(subscription.getChannels()).isEmpty();
}
@Test
public void testPUnsubscribePatternWithChannels() {
byte[][] pattern = new byte[][] { "s*".getBytes() };
subscription.subscribe(new byte[][] { "a".getBytes() });
subscription.pSubscribe(pattern);
subscription.pUnsubscribe(pattern);
verify(asyncCommands).punsubscribe(pattern);
verify(asyncCommands, never()).unsubscribe();
verify(asyncCommands, never()).punsubscribe();
assertThat(subscription.isAlive()).isTrue();
assertThat(subscription.getPatterns().isEmpty()).isTrue();
assertThat(subscription.getPatterns()).isEmpty();
Collection<byte[]> channels = subscription.getChannels();
assertThat(channels.size()).isEqualTo(1);
assertThat(channels).hasSize(1);
assertThat(channels.iterator().next()).isEqualTo("a".getBytes());
}
@Test
public void testUnsubscribePatternWithChannelsSomeLeft() {
byte[][] pattern = new byte[][] { "a*".getBytes() };
subscription.pSubscribe(new byte[][] { "a*".getBytes(), "b*".getBytes() });
subscription.pSubscribe("a*".getBytes(), "b*".getBytes());
subscription.subscribe(new byte[][] { "a".getBytes() });
subscription.pUnsubscribe(pattern);
verify(asyncCommands, never()).unsubscribe();
verify(asyncCommands, never()).punsubscribe();
verify(asyncCommands).punsubscribe(pattern);
assertThat(subscription.isAlive()).isTrue();
Collection<byte[]> channels = subscription.getChannels();
assertThat(channels.size()).isEqualTo(1);
assertThat(channels).hasSize(1);
assertThat(channels.iterator().next()).isEqualTo("a".getBytes());
Collection<byte[]> patterns = subscription.getPatterns();
assertThat(patterns.size()).isEqualTo(1);
assertThat(patterns).hasSize(1);
assertThat(patterns.iterator().next()).isEqualTo("b*".getBytes());
}
@Test
public void testPUnsubscribeAllNoPatterns() {
subscription.subscribe(new byte[][] { "s".getBytes() });
subscription.pUnsubscribe();
verify(asyncCommands, never()).unsubscribe();
verify(asyncCommands, never()).punsubscribe();
assertThat(subscription.isAlive()).isTrue();
assertThat(subscription.getPatterns().isEmpty()).isTrue();
assertThat(subscription.getPatterns()).isEmpty();
Collection<byte[]> channels = subscription.getChannels();
assertThat(channels.size()).isEqualTo(1);
assertThat(channels).hasSize(1);
assertThat(channels.iterator().next()).isEqualTo("s".getBytes());
}
@Test
public void testPUnsubscribeNotAlive() {
subscription.subscribe(new byte[][] { "a".getBytes() });
subscription.unsubscribe();
assertThat(subscription.isAlive()).isFalse();
subscription.pUnsubscribe();
verify(connectionProvider).release(pubsub);
verify(pubsub).removeListener(any(LettuceMessageListener.class));
verify(asyncCommands).unsubscribe();
@@ -304,31 +369,40 @@ public class LettuceSubscriptionTests {
@Test(expected = RedisInvalidSubscriptionException.class)
public void testPSubscribeNotAlive() {
subscription.subscribe(new byte[][] { "a".getBytes() });
subscription.unsubscribe();
assertThat(subscription.isAlive()).isFalse();
subscription.pSubscribe(new byte[][] { "s*".getBytes() });
}
@Test
public void testDoCloseNotSubscribed() {
subscription.doClose();
verify(asyncCommands, never()).unsubscribe();
verify(asyncCommands, never()).punsubscribe();
}
@Test
public void testDoCloseSubscribedChannels() {
subscription.subscribe(new byte[][] { "a".getBytes() });
subscription.doClose();
verify(asyncCommands).unsubscribe();
verify(asyncCommands, never()).punsubscribe();
}
@Test
public void testDoCloseSubscribedPatterns() {
subscription.pSubscribe(new byte[][] { "a*".getBytes() });
subscription.doClose();
verify(asyncCommands, never()).unsubscribe();
verify(asyncCommands).punsubscribe();
}